企业官网建设流程全解析

Pod数量没有变，队列深度却越来越高。第一反应是加副本，但新增worker如果没有加载成功，或者旧worker已经失联却仍占着任务，扩容并不会自动解释积压来自哪里。

观测先要有口径。消息代理的待投递消息、已投递未确认消息，和业务任务表里的待处理、执行中、租约失效，是不同视角。它们可以相互核对，不能直接逐项相加成一个“待处理总数”。本文讨论音频长任务，不把Pod数量当作业务完成数量。

同一个任务可能在两个系统中出现

任务已被worker领取，代理里仍显示未确认；数据库里同时有running记录。这不是两项待办，而是同一任务在两个系统中的投影。重投还可能让同一task_id对应多个消息投递记录。

先在任务表用task_id识别业务任务，另存attempt_id区分执行尝试。执行中记录由哪个Pod实例领取、领取时间、租约到期时间和最后阶段。Pod名称可能重用，跟踪具体实例时应记录Pod UID，执行身份仍以业务attempt为准。

不要把完整task_id、用户ID或原始文件名全部做成监控标签。聚合指标维度保持受控，具体任务定位通过受权限控制的日志或查询页面完成，避免指标基数随任务数量无限增长。

租约失效是待核查状态，不等于可以立即重做

租约过期可能是进程退出、心跳线程受阻，也可能是观测数据迟到。把失效任务单列，结合worker状态与结果提交规则决定恢复。直接把它归入普通pending并重复执行，可能让两个执行同时提交结果。

下面只对同一时刻的任务快照分类。截止时刻到期就视为租约失效；这是一条明确的示例口径，真实系统需统一时钟来源，并为采集延迟定义策略。

def classify_tasks(tasks, now): counts = {"pending": 0, "active": 0, "lease_expired": 0, "terminal": 0, "unknown": 0} for task in tasks: state = task["state"] if state == "pending": key = "pending" elif state == "running": expiry = task.get("lease_until") key = ("unknown" if expiry is None else "active" if expiry > now else "lease_expired") elif state in {"succeeded", "failed", "cancelled"}: key = "terminal" else: key = "unknown" counts[key] += 1 return counts

进入函数前应校验时间字段类型与单位，并确保快照中每个业务任务只有一条当前状态记录。如果输入是执行尝试明细，需要先按既定规则整理，不能用这个函数替你去重。

观察数量，还要观察等待多久

十个短任务与十个长任务的含义不同。任务数、最老待处理任务年龄、领取速率、完成速率、失败分类与阶段耗时应该一起看。这里只定义要观察的量，没有填造实际曲线或阈值。

如果pending增加而领取速率下降，先查消费者是否可接活、队列连接和权限；如果active堆积而完成速率下降，再按解码、模型执行、持久化等阶段定位。lease_expired增加时，查心跳、Pod事件和结果提交冲突。三个方向的处理动作不同。

队列ready与unacknowledged指标能够辅助定位代理层发生了什么；代理没有提供业务成功定义，ack时机还取决于应用协议。因此不能看到消息已确认，就替用户显示音频产物已验收。

告警应包含口径与下一步

一条可操作告警需要说明采集时间、当前哪类任务异常、数据是否新鲜，以及优先检查的环节。若任务表查询本身失败，不应返回零积压，应该标记观测缺失。页面上零与未知必须能分辨。

本地合成快照覆盖待处理、有效租约、到期边界、已结束、缺失租约和未知状态，分类合计与输入数一致。没有连接RabbitMQ、Kubernetes或生产任务表，没有实际积压曲线、扩容实验或告警效果结论。

参考：RabbitMQ监控与消息指标，2026-10-20核对。业务分类、租约口径与告警设计为应用层示例。