← 返回文章列表

2026年6月25日

AiVedio：RocketMQ 可靠异步设计

从 Outbox、Inbox 去重、Topic/Tag/Consumer Group、延迟消息、DLQ、局部顺序、重试补偿、容量估算和高可用降级出发，设计 AI 视频平台的 RocketMQ 可靠异步链路。

第 07 章：RocketMQ 可靠异步设计

本章定位：说明 RocketMQ 如何承载 AI 视频生成、状态轮询、输出回源、媒体处理、最终导出、计费结算和补偿任务，同时明确它不能替代 PostgreSQL 事实源，也不能天然提供业务 exactly-once。

版本说明：以下设计以 RocketMQ 5.x 的概念和语义为主要参考。若项目使用 4.x Remoting 客户端，具体 API、延迟消息配置和消费模型会不同，但“至少一次、业务幂等、可靠发布、状态机校验、补偿对账”的工程原则不变。Go 侧应在项目立项时明确使用 5.x gRPC SDK 还是历史 Remoting SDK，避免两套语义和运维方式混用。

---

7.1 本章要解决的业务问题

AI 视频平台中的核心任务具有五个共同特征：

1. 长耗时：模型生成可能持续数十秒到数十分钟，最终渲染也可能持续很久。
2. 高成本：一次重复提交可能产生真实供应商费用、GPU 成本或重复计费。
3. 强异步：API 请求不能同步等待模型、回源、转码和渲染完成。
4. 弱确定性：第三方可能已经受理请求，但本地因为超时、断网或进程崩溃没有拿到结果。
5. 阶段多：一个生成请求通常会派生提交、轮询、回调、下载、探测、转码、审核、通知、结算等多个步骤。

RocketMQ 在这里解决的不是“把所有状态放进队列”，而是以下问题：

• 将 API 写请求与慢任务执行解耦，避免同步链路被供应商延迟拖垮。
• 在突发流量下削峰，让消费者按照供应商配额、GPU 容量和租户优先级处理任务。
• 为暂时性故障提供重新投递能力。
• 为轮询、超时检查和补偿提供延迟触发能力。
• 让生成、回源、转码、导出、计费和通知拥有独立的消费组、扩缩容策略和故障域。
• 在消费者重启、网络抖动和 ACK 丢失时，仍能依靠业务幂等恢复。

本章的核心判断是：

RocketMQ 是可靠的异步传递层，不是任务、余额和媒体资产的唯一事实源。PostgreSQL 记录“事实”，RocketMQ 传递“需要做什么”或“发生了什么”，Redis 负责限流与短期协调，对象存储负责视频、图片、音频和中间产物。

---

7.2 核心设计原则

原则一：默认接受至少一次，不虚构 exactly-once

RocketMQ 的可靠消费应按“至少一次”建模。生产端超时重试、消费者处理成功但 ACK 丢失、消费者超时后重新投递，都可能产生重复消息。业务上的“只生效一次”必须由以下机制共同完成：

• 稳定的 event_id。
• inbox_dedup 唯一约束。
• 任务状态机与版本号校验。
• 计费流水唯一业务键。
• 第三方提交幂等键或提交不确定态对账。
• 定期补偿与人工审计。

原则二：消息不是数据库行的替代品

MQ 中的消息可能因保留期到期或磁盘策略而被清理。任务状态、供应商任务号、计费流水、重放审计和最终资产信息必须落 PostgreSQL。消费者偏移量只能表示“某消费组读到了哪里”，不能表示业务任务已经正确完成。

原则三：先持久化业务事实，再可靠发布

业务服务在一个 PostgreSQL 事务内同时写入：

• 任务或业务状态。
• Outbox 事件。

事务提交后，由 Outbox Relay 发布到 RocketMQ。这样避免“数据库已提交但消息未发送”的双写裂缝。

原则四：ACK 必须晚于本地持久化

消费者应按以下顺序处理：

接收消息
  → 校验消息
  → 开启数据库事务
  → 插入 inbox_dedup
  → 校验并更新业务状态
  → 写入下游 outbox
  → 提交数据库事务
  → 返回消费成功 ACK

若数据库提交成功但 ACK 丢失，消息会再次投递，inbox_dedup 会将其识别为重复并安全返回成功。反过来，如果先 ACK 再提交数据库，进程崩溃会造成不可恢复的业务丢失。

原则五：区分“消费重试”和“业务调度重试”

• 消费重试：用于短暂数据库故障、网络闪断、进程异常等低概率技术故障。
• 业务调度重试：用于供应商 429、配额不足、模型排队、稍后轮询、输出尚未就绪等预期业务状态。

后者不应长期占用 RocketMQ 的失败重试链路。正确方式通常是：更新任务为 WAITING_RETRY，计算 next_retry_at，发送延迟消息，然后 ACK 原消息。

原则六：不在消息中传输媒体大对象

消息只携带 ID、版本、追踪信息和最小业务参数。原始视频、输出视频、图片、音频、缩略图和大体积供应商响应应放在对象存储或 PostgreSQL 的受控字段中，消息只传 asset_id、object_key 或 payload_ref。

原则七：只在必要范围内使用顺序

绝大多数流程只需要同一 task_id 的局部顺序，甚至可以依靠状态版本和条件更新容忍乱序。全局顺序会把吞吐压缩到少量队列，并放大头阻塞。即使使用 FIFO，数据库状态机仍是最终正确性防线。

原则八：每条异步链路必须有补偿入口

可靠异步不是“消息发出去就结束”。每个阶段都应能回答：

• 消息是否成功发布？
• 消费是否成功落库？
• 第三方是否已经执行？
• 任务长时间不推进时由谁扫描？
• DLQ 中的消息由谁处理？
• 重放会不会重复生成或重复计费？

---

7.3 RocketMQ 在系统中的边界

组件	应承担的职责	不应承担的职责
PostgreSQL	任务事实、状态机、尝试记录、计费流水、Outbox、Inbox、重放审计	高频毫秒级进度广播、视频二进制
RocketMQ	异步解耦、削峰、重新投递、延迟触发、独立消费组、死信隔离	余额唯一事实源、任务唯一状态源、天然 exactly-once
Redis	限流、并发槽位、短期缓存、租约、通知加速	核心任务和计费的唯一持久化
对象存储/CDN	原始素材、供应商输出、代理视频、切片、最终成品	业务状态流转、消费进度

控制面与媒体数据面

RocketMQ 属于控制面：传输任务命令、状态事件和资源引用。

正确：asset_id、object_key、task_id、render_spec_id
错误：把 300 MB 参考视频或完整 MP4 放入消息体

媒体数据面应采用：

客户端 / Worker ↔ 对象存储 ↔ CDN

而不是：

客户端 → Go API → RocketMQ → Worker
                 传输视频二进制

---

7.4 详细架构与组件职责

7.4.1 文字架构图

Browser / Mobile
       │
       ▼
Generation API / Render API / Asset API
       │
       ├── PostgreSQL 业务表
       └── outbox_events             同一事务
                │
                ▼
          Outbox Relay 集群
                │ 同步发送，失败重试
                ▼
        RocketMQ Broker / Proxy
                │
   ┌────────────┼──────────────────────────────┐
   ▼            ▼              ▼               ▼
Scheduler    Poll Worker    Asset Fetch     Render Worker
Consumer     Consumer       Consumer        Consumer
   │            │              │               │
   ▼            ▼              ▼               ▼
Provider     Provider API   Object Storage   FFmpeg/GPU
Adapter      Query          Media Pipeline   Render Farm
   │            │              │               │
   └────────────┴──────┬───────┴───────────────┘
                       ▼
                  PostgreSQL
            inbox_dedup / task_events
            task_attempts / credit_ledger
            downstream outbox_events
                       │
                       ▼
          Notification / Billing / Reconcile

7.4.2 组件职责

1. 业务服务

• 在 PostgreSQL 事务内创建任务、额度预占和 Outbox 事件。
• 不直接依赖“MQ 发送成功后才提交数据库”。
• 对 API 幂等键建立唯一约束。

2. Outbox Relay

• 使用 FOR UPDATE SKIP LOCKED 或租约批量领取待投递事件。
• 将同一 Outbox 事件稳定映射为同一 event_id。
• 同步等待 Broker ACK；超时不代表一定失败，因此允许再次发送同一 event_id。
• 发布成功后更新 published_at，并保留发送次数和最后错误。
• 对长期失败事件报警，不静默丢弃。

3. RocketMQ

• 按业务域和消息类型隔离 Topic。
• 为不同下游使用独立 Consumer Group。
• 负责存储、投递、消费进度、重试和 DLQ。
• 不负责判断任务状态是否允许迁移。

4. 消费者

• 校验消息版本和必填字段。
• 使用 Inbox 去重。
• 使用状态机条件更新防止乱序和竞态。
• 只在本地事实持久化后返回成功。
• 通过有界 worker pool、信号量和供应商并发槽位限制并发。

5. 延迟调度器

• 使用延迟消息触发下一次轮询、重试、超时检查或补偿。
• 收到延迟消息后仍需读取 PostgreSQL 校验 next_retry_at 和当前状态。
• 延迟消息只是“唤醒信号”，不是绝对精确的定时器和唯一事实源。

6. DLQ / Poison Message Processor

• 对超过最大重试次数或结构不可处理的消息进行隔离。
• 记录错误分类、消息摘要、原事件 ID、消费者组、首次和最后失败时间。
• 支持修复后选择性重放。

7. Reconciler

• 扫描长时间停留在 SUBMITTING、PROVIDER_RUNNING、FETCHING、RENDERING 等状态的任务。
• 对照供应商状态、对象存储和计费流水修复遗漏事件。
• 在 MQ 长时间故障、延迟消息丢失或人工误操作时提供最终恢复能力。

---

7.5 Topic、Tag、Consumer Group 与 Message Key 设计

7.5.1 Topic 设计原则

Topic 应按以下维度拆分：

• 业务语义是否不同。
• 消息类型是否不同，例如普通、FIFO、延迟、事务消息。
• SLA 和积压容忍度是否不同。
• 消费者资源是否不同。
• 故障是否需要隔离。

不要按以下方式拆分：

• 每个用户一个 Topic。
• 每个租户一个 Topic。
• 每个任务一个 Topic。
• 仅为了区分一个小字段就创建 Topic。

一个可落地的逻辑划分如下：

逻辑 Topic	典型事件	Consumer Group	关键隔离理由
generation.submit	generation.submit.requested	cg.scheduler.submit.v1	供应商提交，高成本、强限流
generation.poll	generation.poll.due	cg.provider.poll.v1	延迟消息、高频但轻量
generation.callback	provider.callback.received	cg.callback.apply.v1	外部回调乱序与重复处理
asset.fetch	asset.fetch.requested	cg.asset.fetch.v1	网络与对象存储带宽密集
asset.postprocess	asset.probe、asset.transcode、asset.thumbnail	不同处理组	CPU/GPU/IO 资源隔离
render.export	render.export.requested	cg.render.export.v1	长耗时 GPU/FFmpeg 任务
billing.settle	billing.settle.requested	cg.billing.settle.v1	财务语义和审计要求高
notification.event	task.state.changed	cg.notification.push.v1	可降级，不能反压核心链路
compensation.reconcile	task.reconcile.due	cg.reconcile.v1	补偿流量与主链路隔离

在 RocketMQ 5.x 中，Topic 的消息类型是显式属性，因此物理 Topic 可能需要进一步拆成例如：

generation.submit.normal
generation.poll.delay
generation.state.fifo

逻辑命名可以保持业务可读性，但部署时必须与实际版本、消息类型约束一致。

7.5.2 Tag 的使用

Tag 适合区分同一业务域、同一消息类型、相近 SLA 的子事件，例如：

asset.postprocess
  ├── PROBE
  ├── TRANSCODE_PROXY
  ├── THUMBNAIL
  └── WAVEFORM

当不同事件需要完全不同的资源池、保留期或故障隔离时，应拆 Topic，而不是只靠 Tag。

同一 Consumer Group 内所有实例必须保持一致的订阅表达式，否则会出现消费行为不一致。

7.5.3 Consumer Group 设计

• 每个独立业务订阅者使用独立 Consumer Group。
• 同一组内实例执行相同逻辑并共同负载均衡。
• 灰度新版本若要同时观察旧版本消息，应使用独立 shadow group；不要让两个不同逻辑版本共享同一组抢消息。
• Consumer Group 名称应包含业务、用途和兼容版本，不包含实例 ID。

示例：

cg.generation.scheduler.v1
cg.asset.fetch.v2
cg.billing.settle.v1
cg.notification.websocket.v1

7.5.4 Message Key 与消息组

建议：

RocketMQ Keys     = event_id
业务属性 task_id   = aggregate_id
trace_id          = 分布式追踪
message_group     = task_id，仅在确需 FIFO 时设置

event_id 应全局唯一且在 Outbox 重发时保持不变，便于查询、去重和追踪。task_id 用于聚合同一任务的事件，但不能替代唯一事件 ID。

---

7.6 关键消息结构与数据库表

7.6.1 通用消息信封

{
  "event_id": "0198f6d8-5ab1-7e3a-9d55-7c0d8b102001",
  "event_type": "generation.submit.requested",
  "schema_version": 3,
  "producer": "generation-service",
  "occurred_at": "2026-06-24T02:30:00Z",
  "aggregate_type": "generation_task",
  "aggregate_id": "task_01J...",
  "tenant_id": "tenant_01J...",
  "trace_id": "7d9f...",
  "correlation_id": "request_01J...",
  "causation_id": "api_command_01J...",
  "payload": {
    "task_id": "task_01J...",
    "task_version": 5,
    "attempt_id": "attempt_01J...",
    "model": "video-model-x",
    "input_asset_ids": ["asset_01J..."],
    "request_ref": "generation_request/task_01J.../v5"
  }
}

设计要求：

• event_id：消息幂等主键。
• event_type：稳定的业务语义，不使用模糊的 do_task。
• schema_version：支持兼容演进。
• aggregate_id：任务、资产或渲染作业 ID。
• task_version：辅助识别过期消息。
• payload：只放最小必要数据，大对象使用引用。
• 不信任消息中的租户权限声明；消费者仍需从数据库校验资源归属。

7.6.2 Outbox 表

CREATE TABLE outbox_events (
    id                bigserial PRIMARY KEY,
    event_id          uuid        NOT NULL,
    aggregate_type    text        NOT NULL,
    aggregate_id      uuid        NOT NULL,
    event_type        text        NOT NULL,
    topic             text        NOT NULL,
    tag               text,
    message_key       text        NOT NULL,
    schema_version    integer     NOT NULL,
    payload           jsonb       NOT NULL,
    occurred_at       timestamptz NOT NULL,
    available_at      timestamptz NOT NULL DEFAULT now(),
    published_at      timestamptz,
    publish_attempts  integer     NOT NULL DEFAULT 0,
    lease_owner       text,
    lease_until       timestamptz,
    last_error_code   text,
    last_error        text,
    created_at        timestamptz NOT NULL DEFAULT now(),
    UNIQUE (event_id)
);

CREATE INDEX idx_outbox_pending
ON outbox_events (available_at, id)
WHERE published_at IS NULL;

关键点：

• event_id 在重发时不变。
• available_at 支持本地延迟投递和失败退避。
• 使用租约避免多个 Relay 重复领取，但即便重复发送也必须安全。
• 已发布数据按时间归档或分区，避免表和索引无限膨胀。

7.6.3 Inbox 去重表

CREATE TABLE inbox_dedup (
    consumer_name     text        NOT NULL,
    event_id          uuid        NOT NULL,
    aggregate_id      uuid,
    processed_at      timestamptz NOT NULL DEFAULT now(),
    result_code       text        NOT NULL,
    payload_hash      bytea,
    PRIMARY KEY (consumer_name, event_id)
);

同一事件可以被不同 Consumer Group 独立处理，因此唯一键必须包含 consumer_name。

对于超高吞吐消费者，可以按月分区并根据消息最大重放窗口设置清理周期；不能在仍可能重放时过早删除去重记录。

7.6.4 任务尝试表

CREATE TABLE task_attempts (
    attempt_id                uuid PRIMARY KEY,
    task_id                   uuid        NOT NULL,
    attempt_no                integer     NOT NULL,
    provider                  text        NOT NULL,
    provider_idempotency_key  text,
    provider_job_id           text,
    status                    text        NOT NULL,
    submit_started_at         timestamptz,
    submit_finished_at        timestamptz,
    next_retry_at             timestamptz,
    error_class               text,
    error_code                text,
    error_message             text,
    created_at                timestamptz NOT NULL DEFAULT now(),
    updated_at                timestamptz NOT NULL DEFAULT now(),
    UNIQUE (task_id, attempt_no),
    UNIQUE (provider, provider_job_id)
);

provider_job_id 允许为空，因为“供应商已受理、本地超时”的窗口内可能尚未获取任务号。

7.6.5 Poison Message 与人工重放表

CREATE TABLE message_quarantine (
    id                 bigserial PRIMARY KEY,
    consumer_name      text        NOT NULL,
    original_event_id  uuid,
    topic              text        NOT NULL,
    message_key        text,
    schema_version     integer,
    payload_hash       bytea       NOT NULL,
    payload_snapshot   jsonb,
    error_class        text        NOT NULL,
    error_message      text        NOT NULL,
    first_failed_at    timestamptz NOT NULL,
    last_failed_at     timestamptz NOT NULL,
    delivery_count     integer     NOT NULL,
    status             text        NOT NULL,
    UNIQUE (consumer_name, original_event_id, payload_hash)
);

CREATE TABLE message_replays (
    replay_id           uuid PRIMARY KEY,
    original_event_id   uuid        NOT NULL,
    replay_event_id     uuid        NOT NULL UNIQUE,
    requested_by        text        NOT NULL,
    reason              text        NOT NULL,
    ticket_no           text,
    dry_run_result      jsonb,
    status              text        NOT NULL,
    created_at          timestamptz NOT NULL DEFAULT now(),
    executed_at         timestamptz
);

人工重放必须生成新的 replay_event_id，同时保留 original_event_id。业务副作用仍使用原始业务键去重，不能因为换了消息 ID 就重复生成或重复扣费。

---

7.7 文字版时序图

7.7.1 正常可靠发布与消费

1. Browser 调用 Generation API，携带 Idempotency-Key。
2. Generation Service 开启 PostgreSQL 事务。
3. 创建 generation_tasks，预占额度，写入 outbox_events。
4. PostgreSQL 提交事务，API 返回 task_id 和 QUEUED。
5. Outbox Relay 领取未发布事件并发送 generation.submit。
6. RocketMQ 持久化消息并返回 ACK。
7. Relay 将 outbox_events.published_at 更新为当前时间。
8. Scheduler Consumer 收到消息。
9. Consumer 在事务内插入 inbox_dedup，并将任务 QUEUED → SUBMITTING。
10. Consumer 创建 task_attempt，提交事务。
11. Consumer 在事务外调用 Provider Adapter。
12. 供应商返回 provider_job_id。
13. Consumer 在新事务内写入 provider_job_id，将任务改为 PROVIDER_RUNNING，
    同时写入 generation.poll 的延迟 Outbox 事件。
14. 数据库提交后，Consumer 向 RocketMQ 返回成功。
15. 后续 Poll Consumer、Asset Fetch Consumer、Media Worker 按相同模式推进任务。

7.7.2 “供应商已受理，但本地超时”时序

1. Consumer 已创建 attempt_id 和稳定 provider_idempotency_key。
2. Consumer 调用供应商提交接口。
3. 供应商创建了远端任务，但响应在网络中丢失。
4. 本地请求超时，不能直接判断为失败，也不能立即创建新 attempt 重试。
5. 本地将 attempt 标记为 SUBMIT_UNKNOWN，任务进入 RECONCILING。
6. 发送延迟 reconcile 消息并 ACK 当前消息。
7. Reconciler 使用 provider_idempotency_key、client_request_id、时间窗口或供应商查询接口查找远端任务。
8. 若找到远端任务，回填 provider_job_id，继续轮询。
9. 若供应商明确确认未创建，才允许重试同一幂等键或创建新的受控 attempt。
10. 若无法确认，则根据成本、SLA 和最大不确定窗口进入人工处理或有限重试，不能无限自动重提。

---

7.8 正常流程

7.8.1 生成提交

1. API 创建任务和 Outbox。
2. Relay 发布 generation.submit.requested。
3. Scheduler Consumer 校验租户、任务状态和版本。
4. 获取 Redis 中用户、租户、模型、供应商并发槽位。
5. 创建 task_attempts 并调用供应商。
6. 成功后写入 provider_job_id，释放“提交中”租约，进入 PROVIDER_RUNNING。
7. 写入下一次轮询延迟事件。

注意：并发槽位是准入和加速机制，任务事实仍在 PostgreSQL。Redis 计数泄漏由租期和对账修复。

7.8.2 状态轮询

1. 延迟消息在 next_poll_at 附近触发。
2. Poll Consumer 读取任务当前状态。
3. 若任务已完成、失败或取消，消息视为过期并直接 ACK。
4. 若尚未到 next_poll_at，重新发送剩余延迟或交由 DB 调度器处理。
5. 调用供应商查询接口。
6. 未完成：更新进度和下一次轮询时间，写入新延迟事件。
7. 已完成：写入输出引用，发布 asset.fetch.requested。

进度更新不应每秒写 PostgreSQL。可将高频进度放 Redis，按阈值或时间间隔合并落库。

7.8.3 输出回源与媒体处理

1. Fetch Consumer 领取任务，验证供应商输出 URL 尚有效。
2. 采用流式下载写入对象存储，不将整个文件读入内存。
3. 计算 checksum、执行 ffprobe、写入资产记录。
4. 发布探测、转码、缩略图、波形和内容复审事件。
5. 各媒体消费者使用独立组和资源池。
6. 所有必需产物完成后，将任务推进为 SUCCEEDED，触发通知和结算。

7.8.4 最终导出

1. Render API 固化 project_revision_id 和 render_spec_id。
2. 事务内创建 render_jobs 与 Outbox。
3. Render Consumer 读取不可变时间轴版本，生成 Render DAG。
4. 执行 FFmpeg/GPU 渲染并将结果写入对象存储。
5. 通过唯一 render_job_id 防止重复导出副作用。
6. 完成后发布通知与计费结算事件。

---

7.9 异常流程与竞态条件

场景	风险	正确处理
DB 已提交，Relay 发送前崩溃	消息未立即发出	Outbox 仍为未发布，其他 Relay 继续发送
Broker 已收消息，Relay 本地超时	同一消息可能重复发布	保持相同 event_id 重发，消费者去重
Consumer DB 提交后、ACK 前崩溃	重复消费	Inbox 唯一约束返回已处理
Consumer ACK 后才尝试提交 DB	业务永久丢失	禁止；必须先提交再 ACK
消费耗时超过可见/处理超时	同一消息并发执行	缩短单次 handler、配置合理超时、使用 attempt 状态和租约防重入
延迟轮询消息重复或乱序	旧轮询覆盖新状态	读取当前状态和 next_poll_at，旧消息变成 no-op
取消与供应商完成同时发生	任务终态冲突	条件更新、版本号、终态优先规则和补偿退款
FIFO 消息中的一条持续失败	后续同组消息被阻塞	有限重试后隔离，避免超大 message group
Poison Message 无限重试	重试风暴、积压	结构错误直接隔离，业务不可恢复错误不进入技术重试
Consumer 扩容或重平衡	短时重复和抖动	处理逻辑幂等，优雅停机，限制单实例预取和在途量
DLQ 人工重放旧命令	重复生成、重复扣费	重放前 dry-run，校验当前状态和业务唯一键
消息保留期内未消费完	历史消息被清理	监控 lag 与最老消息年龄，容量预留，必要时暂停入口并从 DB 补偿
Schema 升级不兼容	新老消费者解析失败	向后兼容字段、schema_version、双读/双写窗口
供应商已受理但本地超时	重复远端任务和成本	SUBMIT_UNKNOWN + 查询/对账，不盲目重提
结算消息重复	重复扣费	UNIQUE(ledger_business_type, business_id)

7.9.1 取消与完成竞态

推荐状态规则：

PROVIDER_RUNNING + cancel_requested
    ├── 供应商确认取消 → CANCELED
    ├── 供应商已完成   → SUCCEEDED 或 CANCELED_WITH_OUTPUT，按产品规则决定
    └── 结果不确定     → CANCEL_RECONCILING

不能简单采用“最后一条消息覆盖前一条消息”。所有状态迁移应携带当前版本或允许的前置状态：

UPDATE generation_tasks
SET status = 'SUCCEEDED', version = version + 1
WHERE task_id = $1
  AND status IN ('PROVIDER_RUNNING', 'FETCHING')
  AND version = $2;

更新行数为 0 时，消费者应重新读取状态并判定消息是否过期，而不是无条件重试。

---

7.10 幂等、一致性、重试与补偿设计

7.10.1 四层幂等

第一层：API 幂等

UNIQUE(tenant_id, client_idempotency_key)

防止用户双击、客户端重试或网关超时造成重复任务。

第二层：消息幂等

PRIMARY KEY(consumer_name, event_id)

防止重复发布和重复投递造成本地状态重复应用。

第三层：状态机幂等

通过任务状态、版本号和唯一约束，确保旧消息、乱序消息和重复消息只能成为 no-op。

第四层：外部副作用幂等

• 供应商支持幂等键：使用稳定 provider_idempotency_key。
• 供应商不支持幂等键：持久化 attempt，提交超时进入不确定态，先查询再重试。
• 对象存储：使用确定性 object key 和 checksum，重复上传可覆盖同内容或检测冲突。
• 计费：以业务类型和业务 ID 建唯一流水。
• 通知：允许至少一次，前端按 event_id 或 task_version 去重。

7.10.2 消费事务模板

// 伪代码：展示事务边界，不代表完整 SDK API。
func handleEvent(ctx context.Context, msg Event) error {
    return db.WithTx(ctx, func(tx Tx) error {
        inserted, err := tx.InsertInboxIfAbsent(
            "cg.asset.fetch.v1", msg.EventID, msg.AggregateID,
        )
        if err != nil {
            return Retryable(err)
        }
        if !inserted {
            return nil // 已处理，安全 ACK
        }

        task, err := tx.LoadTaskForUpdate(msg.AggregateID)
        if err != nil {
            return Retryable(err)
        }
        if !task.CanApply(msg.EventType, msg.TaskVersion) {
            return tx.MarkInboxResult(msg.EventID, "STALE_NOOP")
        }

        if err := tx.ApplyTransition(task, msg); err != nil {
            return err
        }
        if err := tx.InsertDownstreamOutbox(task.NextEvent()); err != nil {
            return err
        }
        return tx.MarkInboxResult(msg.EventID, "APPLIED")
    })
}

注意：外部 HTTP 调用不应放在上述数据库事务内。应先持久化 attempt 和状态，再在事务外调用，最后用新事务记录结果。

7.10.3 错误分类与重试矩阵

错误类别	示例	MQ 消费重试	业务延迟重试	是否可能重复生成/计费
瞬时基础设施错误	DB 短暂断连、网络闪断	是，有限次数	可选	通常不会，但仍需幂等
资源压力	消费者过载、供应商并发满	不应靠失败重试限流	是	若已产生外部调用则有风险
供应商 429/503	限流、临时不可用	原消息应 ACK	是，退避+jitter	重提之前检查 attempt 状态
参数错误	不支持的分辨率、缺少素材	否	否	不应产生费用
内容审核拒绝	prompt 或输出违规	否	通常否	不应盲目换供应商绕过审核
Schema 不支持	新版本字段无法解析	否，直接隔离	修复后人工重放	取决于是否已执行副作用
Provider Submit 超时	对方可能已创建任务	否，不能直接重提	先对账再决定	高风险
输出 URL 暂不可用	远端任务完成但文件未就绪	原消息 ACK	是	不应重新生成
对象存储写超时	可能已写成功	有限技术重试	可选	用确定性 key/checksum 防重复
计费写入冲突	重复结算消息	否，唯一键识别为成功	否	唯一约束防重复扣费

7.10.4 退避策略

业务延迟重试建议：

base = 5s
cap  = 10m
next = min(cap, base * 2^attempt) + random_jitter

但不是所有场景都从 5 秒开始：

• 模型状态轮询可依据供应商建议动态调整。
• 429 应尊重 Retry-After。
• 输出文件刚生成时可采用短间隔，随后逐步拉长。
• 超过最大等待时间后进入明确失败或人工处理，不能无限轮询。

7.10.5 补偿与对账

建议至少有以下扫描器：

1. Outbox 补偿：扫描长期未发布事件。
2. SUBMITTING 补偿：扫描提交开始后超时但无明确结果的 attempt。
3. PROVIDER_RUNNING 补偿：扫描长时间没有轮询推进的任务。
4. FETCHING 补偿：检查对象存储是否已有文件但数据库未更新。
5. MEDIA_PROCESSING 补偿：检查缺失的必需派生产物。
6. BILLING 补偿：对照任务终态、供应商账单和 credit_ledger。
7. Redis 槽位对账：根据 PostgreSQL 在途任务重建并发计数。

补偿任务也要幂等，并记录每次修复动作，不能成为另一条不可观测的“黑盒定时任务”。

---

7.11 局部顺序、全局顺序与乱序容忍

7.11.1 为什么通常不需要全局顺序

任务 A 和任务 B 之间没有业务先后关系。如果把所有事件放入全局有序队列：

• 任一慢任务都会阻塞其他任务。
• 消费并发显著下降。
• 单个热点任务可能形成头阻塞。
• 扩容能力受限。

因此通常只考虑同一 task_id、render_job_id 或 project_id 的局部顺序。

7.11.2 什么时候使用 FIFO

适合：

• 同一任务的状态事件必须按版本发布给某个不具备乱序处理能力的下游。
• 同一项目版本的串行变更应用。
• 同一资产的不可并行元数据更新。

不适合：

• 仅仅因为“看起来更安全”。
• 消费者本身存在外部回调、数据库写入和人工操作等其他并发来源。
• 一个 message group 会聚合海量消息。

RocketMQ 的 FIFO 依赖消息组；不同消息组仍可并行。同组消息失败重试时，后续消息可能被阻塞。因此即使使用 FIFO，也要限制重试次数并准备隔离策略。

7.11.3 推荐的双保险

MQ：尽量提供同 task_id 的局部顺序
DB：使用 version、前置状态和唯一约束验证迁移

MQ 顺序用于降低乱序概率，数据库状态机用于保证最终正确性。

---

7.12 延迟消息、死信队列、Poison Message 与人工重放

7.12.1 延迟消息的正确用途

适合：

• 下一次供应商状态轮询。
• 429/503 后的退避重试。
• 任务超时检查。
• 临时 URL 到期前的回源提醒。
• 补偿任务的延迟触发。

不适合：

• 将精确金融结算时间只托付给延迟消息。
• 把所有定时任务都迁入 MQ，完全取消数据库扫描。
• 依赖一条延迟消息作为唯一恢复入口。

每条延迟消息必须携带或可查询：

aggregate_id
expected_status
expected_version
not_before
reason
retry_attempt

收到后先校验数据库当前状态，过期消息直接 no-op。

7.12.2 死信队列

达到最大消费重试次数后，消息进入 DLQ。DLQ 不是“自动修复”，而是隔离区。生产要求：

• 每个核心 Consumer Group 都有 DLQ 指标和报警。
• 明确负责人和处理时限。
• 自动提取消息元数据进入 message_quarantine。
• 区分代码缺陷、数据缺陷、权限问题、依赖故障和历史脏数据。
• 修复后选择性重放，而不是一键全量重放。

7.12.3 Poison Message 判定

以下情况不应重复消耗大量重试预算：

• JSON 无法解析。
• 缺少必填字段。
• schema_version 不受支持。
• 业务不变量永久不满足。
• 载荷超过允许大小。
• 引用的租户或资源明显非法。

处理方式：

写入隔离记录 → 记录原因与哈希 → 返回成功 ACK → 报警

写隔离记录与 ACK 之间仍然不是分布式事务，因此隔离表必须幂等。若 ACK 丢失，再次投递时命中唯一约束后继续 ACK。

7.12.4 人工重放

推荐流程：

选择原消息
  → 展示当前任务状态和历史副作用
  → dry-run 判断是否仍可执行
  → 审批与审计
  → 生成 replay_event_id
  → 保留 original_event_id
  → 发布到专用 replay Topic 或原 Topic
  → 消费端再次执行业务幂等校验

不推荐直接重置整个 Consumer Group Offset，因为这会把大量已成功消息重新送入所有副作用链路，可能造成重复生成、重复通知、重复导出和计费压力。

---

7.13 性能瓶颈与容量估算

7.13.1 消息到达速率

定义：

λ_task_i  = 第 i 类业务任务到达速率
fanout_i  = 每个任务平均产生的 MQ 消息数
λ_msg     = Σ(λ_task_i × fanout_i)

AI 视频任务的消息放大主要来自轮询。若一个任务平均轮询 12 次，再加提交、回源、3 个媒体处理、通知和结算，一个任务产生 20 条消息并不夸张。

示例假设：

峰值生成请求：30 task/s
平均消息放大：20 message/task
峰值消息速率：30 × 20 = 600 message/s

必须按“消息数”而不是只按“用户请求数”规划 MQ。

7.13.2 消费能力

近似公式：

单实例吞吐 μ_instance ≈ concurrency / 平均处理时间
集群吞吐 μ_total       ≈ instance_count × μ_instance × 有效利用率

若 Asset Fetch 平均耗时 20 秒，单实例并发 40，有效利用率按 0.7：

μ_instance ≈ 40 / 20 × 0.7 = 1.4 task/s

如果峰值到达 14 task/s，至少需要约 10 个同等实例，再加故障和突发余量。此类消费者的瓶颈通常是外网带宽、对象存储写吞吐和临时磁盘，而不是 CPU。

7.13.3 积压清空时间

T_drain = backlog / (μ_total - λ_in)

前提是 μ_total > λ_in。例如积压 1,000,000 条，恢复后消费 900 msg/s，新流量仍为 600 msg/s：

T_drain = 1,000,000 / (900 - 600)
        ≈ 3,333 秒
        ≈ 55.6 分钟

只看当前消费 TPS 不够，还要看净清空速度。

7.13.4 存储容量

近似公式：

logical_bytes = λ_avg × avg_message_bytes × retention_seconds
physical_bytes ≈ logical_bytes × 副本系数 × 索引/文件/安全余量

示例：

平均消息速率：160 msg/s
平均消息体：2 KiB
保留时间：72 h
逻辑消息体：约 79 GiB
若按 3 份存储并乘 1.3 余量：约 308 GiB

实际还要考虑 CommitLog、ConsumeQueue、索引、重试 Topic、DLQ、突发峰值和磁盘水位，因此不能只按消息体相乘后贴线部署。

7.13.5 重点监控指标

至少监控：

• Producer 发送成功率、错误率和发送耗时分位数。
• Outbox 未发布数量和最老事件年龄。
• 每个 Topic / Consumer Group 的 lag、最老消息年龄和消费 TPS。
• 消费处理耗时、失败率、重试次数和 DLQ 增量。
• 客户端本地缓存消息数与字节数。
• Broker 磁盘使用率、存储保留窗口和节点健康。
• 按事件类型统计的端到端延迟。
• SUBMIT_UNKNOWN、RECONCILING 数量。
• 重放数量、重放成功率和重放产生的真实副作用。

7.13.6 常见性能瓶颈

1. 轮询放大：任务量不高，但每个任务产生大量 Poll 消息。
2. 热点 message group：所有租户错误地使用同一个顺序键。
3. 消费者内部无界 goroutine：看似吞吐上升，实际导致内存、连接池和供应商限流崩溃。
4. 数据库 Inbox 热点：索引膨胀或单表写入过高，需要分区和批量清理。
5. Outbox 扫描低效：没有 partial index，反复全表扫描。
6. 大消息：携带完整 prompt 历史、供应商响应甚至媒体内容。
7. 通知反压核心 Topic：WebSocket 下游慢导致生成链路受影响，应独立 Topic 和消费组。

---

7.14 高可用与降级方式

7.14.1 Broker 与网络层

• Broker、NameServer/Proxy 等组件跨故障域部署，采用与所选版本匹配的主从切换或控制器方案。
• Producer 配置多个可用端点，发送重试有上限。
• 消费者无状态部署多个副本，避免单实例成为唯一处理者。
• 保留足够磁盘和告警水位，不能等磁盘接近耗尽才处理积压。

7.14.2 Producer 降级

当 RocketMQ 短时不可用：

• API 仍可在 PostgreSQL 中创建任务与 Outbox。
• 用户看到任务为 QUEUED 或 ACCEPTED，但系统应展示排队延迟。
• Relay 持续重试，且监控 Outbox 最老年龄。

当故障持续且 Outbox 快速增长：

• 降低免费用户和低优先级租户入口速率。
• 暂停高成本生成和最终导出新请求。
• 保留状态查询、项目编辑和已完成视频播放。
• 明确返回“系统繁忙”，而不是继续无限接单导致恢复时间失控。

7.14.3 Consumer 降级

• 通知服务故障：核心任务继续，前端退化为轮询。
• 缩略图或波形故障：主视频可先进入“部分就绪”，后台补产物。
• 非关键内容增强失败：跳过增强，使用原 prompt，但必须符合产品规则。
• 某供应商故障：调度器熔断并切换供应商；切换前评估模型能力差异和重复提交风险。
• 延迟消息链路异常：数据库 Reconciler 按 next_retry_at 扫描兜底。

7.14.4 优雅停机

Go Consumer 收到终止信号后：

1. 停止拉取新消息。
2. 等待有界时间内的在途 handler 完成。
3. 未完成消息不提前 ACK。
4. 关闭 Producer、Consumer、数据库和追踪连接。
5. 对正在执行外部提交的 attempt 保留可恢复状态。

不能在消息 listener 中把任务丢给后台 goroutine 后立即返回成功，否则进程退出时 MQ 会认为消息已完成，而后台任务可能尚未落库。

---

7.15 安全风险

7.15.1 访问控制

• RocketMQ 端口只暴露在受控私网。
• 启用并正确配置 ACL；未配置身份验证时，能够访问端口的实体可能执行发送、消费或管理操作。
• Producer、Consumer 和运维账号分离，按 Topic/Group 最小授权。
• 凭据存放在 Secret Manager/KMS，不写入代码、镜像或日志。
• 对管理控制台、重放工具和消费位点操作启用强认证与审计。

7.15.2 数据最小化

消息中避免包含：

• 完整用户提示词历史。
• 可识别个人身份的信息。
• 供应商密钥。
• 预签名 URL 的长期有效凭据。
• 原始媒体二进制。

必要字段应加密或使用短期引用。日志只记录 event_id、task_id、错误码和摘要，不打印完整 payload。

7.15.3 多租户隔离

消息中的 tenant_id 只能用于路由和追踪，不能作为授权的唯一依据。消费者读取任务时必须校验：

task.tenant_id == message.tenant_id
asset.tenant_id == task.tenant_id

发现不一致时进入安全隔离和报警，而不是自动修正。

7.15.4 重放与运维风险

人工重放、跳过 DLQ、修改 Offset 和重新发布计费消息都属于高风险操作，应具备：

• 工单号。
• 操作人。
• dry-run 结果。
• 影响任务列表。
• 审批记录。
• 执行前后快照。

---

7.16 Outbox 与 RocketMQ 事务消息的比较

维度	PostgreSQL Outbox	RocketMQ 事务消息
核心机制	本地事务写业务表和 Outbox，Relay 后发消息	先发 Half Message，再执行本地事务并提交/回滚，Broker 可回查
与业务数据库一致性	直接依赖同一数据库事务，模型直观	依赖事务生产者和 Transaction Checker 正确实现
消息发布延迟	有 Relay 扫描/CDC 延迟	通常更直接
运维可见性	Outbox 行可查询、审计、补发	需要理解 Half Message、回查和生产者可用性
技术耦合	较低，可更换 MQ	更依赖 RocketMQ 事务语义
表膨胀	需要分区、清理和索引治理	无 Outbox 表，但仍需业务审计数据
消费端幂等	仍然必须	仍然必须
外部供应商副作用	不能自动解决	同样不能解决
适合场景	PostgreSQL 是事实源、事件多、需审计重放	本地事务简单、团队熟悉事务消息、希望减少 Relay

RocketMQ 事务消息解决的是“本地事务与消息是否最终一致”，并不保证下游消费结果与上游完全一致，也不消除重复消费。它还要求生产者能够在 Broker 回查时可靠判断本地事务状态。

本项目建议

• 默认使用 Outbox：任务、计费、资产和项目版本本来就以 PostgreSQL 为事实源，Outbox 更容易审计、重放和补偿。
• 有限场景评估事务消息：例如单一、简单、本地事务状态易查询的高吞吐事件。
• 不要对同一个业务事件同时使用“Outbox + 事务消息”做双重发布，这会增加重复和排障复杂度。
• 无论选择哪一种，消费者幂等和第三方提交不确定态处理都不能省略。

---

7.17 常见错误设计及其后果

错误一：数据库提交后直接调用 MQ，没有 Outbox

**后果：**数据库成功、MQ 失败时任务永远停在队列前；补偿只能靠人工猜测。

错误二：认为 RocketMQ 天然 exactly-once

**后果：**重复消息导致重复生成、重复下载、重复导出或重复扣费。

错误三：Consumer 收到消息后先 ACK，再异步处理

**后果：**进程崩溃后消息不会重投，任务永久丢失。

错误四：Provider Submit 超时就直接重新提交

**后果：**供应商可能已经创建任务，平台产生两份视频和两笔成本。

错误五：用 MQ 消费失败充当限流

**后果：**大量消息进入重试链路，形成重试风暴，并掩盖真正故障。

错误六：所有业务共用一个 Topic

**后果：**通知积压可能影响生成，媒体任务挤压计费，无法独立扩容和设定 SLA。

错误七：每个租户或任务创建 Topic

**后果：**元数据膨胀、运维复杂、权限和监控难以管理。

错误八：为“安全”启用全局顺序

**后果：**吞吐下降，单个慢消息阻塞整个系统，扩容收益很低。

错误九：把视频或完整供应商响应放入消息体

**后果：**网络、Broker 磁盘、客户端内存和重试成本迅速放大。

错误十：DLQ 只建不管

**后果：**消息从主链路“消失”，问题延迟到用户投诉才暴露。

错误十一：通过重置 Consumer Offset 做人工重放

**后果：**大批已成功消息被重新执行，造成副作用风暴。

错误十二：把 MQ Offset 当成任务状态

**后果：**无法解释业务是否真正完成，也无法处理外部调用不确定性。

---

7.18 面试官可能追问的 10 个问题与资深回答

问题 1：为什么这个系统需要 RocketMQ，直接用数据库任务表不行吗？

资深回答：

数据库任务表可以做小规模轮询队列，但当生成、轮询、回源、转码、渲染、通知和计费具有不同吞吐、资源和 SLA 时，单表扫描会产生锁竞争、扫描放大和耦合。RocketMQ 提供异步解耦、独立消费组、削峰、重试、延迟和 DLQ，更适合多阶段工作流。但我不会把数据库任务表删掉，因为 PostgreSQL 仍是状态事实源和补偿入口。两者不是替代关系，而是事实层与传递层的分工。

问题 2：如何保证“数据库创建任务”和“发送消息”不丢一个？

资深回答：

使用 Transactional Outbox。在同一个 PostgreSQL 事务中写任务和 Outbox，提交后由 Relay 投递。若 Relay 崩溃，未发布行仍在；若 Broker 已收到但 Relay 超时，Relay 会用同一 event_id 重发，消费者通过 Inbox 去重。这样把不可控的双写问题转化为可接受的重复投递问题。

问题 3：Consumer 执行成功但 ACK 丢了怎么办？

资深回答：

消息会重新投递，所以消费逻辑必须幂等。我在数据库事务内先插入 inbox_dedup，再更新业务状态并写下游 Outbox，事务提交后才 ACK。第二次收到同一 event_id 时，唯一约束命中，直接返回成功，不再执行副作用。

问题 4：第三方已经受理生成，但本地 HTTP 超时，怎么避免重复生成？

资深回答：

这是最关键的边界。提交前先持久化 attempt_id 和稳定的供应商幂等键。超时后不能标记为普通失败并立即重提，而是进入 SUBMIT_UNKNOWN。优先通过幂等键、client request ID 或供应商查询接口找回远端任务；只有确认未创建后才重试。供应商不支持查询时，要设置不确定窗口、成本上限和人工处理策略。MQ 重投也必须先读 attempt 状态，不能再次直接调用。

问题 5：为什么不能说 RocketMQ 保证 exactly-once？

资深回答：

因为发送超时重试、消费处理完成但 ACK 丢失、可见超时和消费者崩溃都会导致再次投递。MQ 能帮助提高投递可靠性，但业务“只生效一次”还涉及数据库、对象存储、供应商和计费系统。真正可落地的是“至少一次传递 + 幂等副作用 + 唯一约束 + 状态机 + 对账”，从业务结果上逼近 effectively-once。

问题 6：任务状态事件要不要使用顺序消息？

资深回答：

通常只需要同一 task_id 的局部顺序，不需要全局顺序。若某下游无法容忍乱序，可以用 task_id 作为 message group。但我仍会在数据库中用任务版本和允许的前置状态做条件更新，因为回调、人工取消和直接数据库补偿并不一定经过同一顺序队列。顺序消息降低乱序概率，状态机保证最终正确性。

问题 7：供应商 429 应让 Consumer 返回失败触发 MQ 重试吗？

资深回答：

不建议长期这样做。429 是预期的业务限流，不应消耗技术重试预算。Consumer 应记录 attempt、计算退避时间、写入延迟消息或 next_retry_at，然后 ACK 原消息。这样能控制节奏、加入 jitter，并避免大量消息挤进 Retry Topic。只有短暂、低概率的基础设施错误才适合使用 MQ 自动消费重试。

问题 8：DLQ 中的消息如何安全重放？

资深回答：

先分类和 dry-run，不直接全量重放。重放记录包含操作人、原因、工单、原事件 ID 和新的 replay ID。消费端根据当前任务状态、原业务幂等键和历史副作用判断：执行、跳过或人工处理。计费和供应商提交尤其不能仅凭新消息 ID 再做一次。重放操作本身必须可审计。

问题 9：如何处理消息积压？

资深回答：

先判断是输入突发、下游变慢、单条 Poison Message、热点顺序组还是消费者容量不足。监控 lag、最老消息年龄、消费耗时和失败率。扩容前先确认供应商配额、数据库连接池、对象存储带宽等下游是否承受得住。计算净清空速度 μ-λ 和预计清空时间。若积压威胁消息保留窗口，应限流入口、暂停低优先级任务并提高关键消费者资源，而不是盲目加 goroutine。

问题 10：Outbox 和 RocketMQ 事务消息怎么选？

资深回答：

两者都解决本地事务与消息发布的最终一致性。这个平台中 PostgreSQL 是任务和计费事实源，且需要审计、补发和多阶段事件，因此我默认选 Outbox。事务消息可以减少 Relay，但依赖 Half Message、二阶段确认和 Transaction Checker，仍然要求消费幂等，也解决不了第三方生成提交的不确定性。团队若已有成熟事务消息运维能力，可在局部简单场景使用，不必全局统一成一种绝对答案。

---

7.19 三分钟口述稿

在这个 AI 视频平台里，我把 RocketMQ 定位为可靠异步传递层，而不是任务事实源。因为生成、回源、转码和导出都是长耗时、高成本任务，API 不能同步等待，所以请求进入系统后，Generation Service 会在同一个 PostgreSQL 事务里完成幂等任务创建、额度预占和 Outbox 写入。事务提交后，Outbox Relay 再把事件投递到 RocketMQ。

这样可以解决数据库和 MQ 双写不一致。Relay 如果发送超时，不能确定 Broker 是否已经收到，所以会用同一个 event_id 重发。RocketMQ 按至少一次消费建模，Consumer 侧通过 inbox_dedup 唯一约束、任务状态机版本和计费业务唯一键保证重复消息不会重复生效。消费顺序是先提交数据库，再返回 ACK；如果提交后 ACK 丢失，下一次投递只会命中去重记录。

Topic 我会按生成提交、状态轮询、输出回源、媒体处理、最终渲染、计费和通知拆分，因为这些任务的资源、SLA 和故障域不同。消息只传 task_id、asset_id、版本和追踪信息，不传视频二进制。顺序方面通常只需要同 task_id 的局部顺序，而且即使使用 FIFO，数据库状态机仍然负责最终正确性。

重试需要分两类。数据库闪断这类低概率技术故障可以走 MQ 消费重试；供应商 429、下一次轮询和输出尚未就绪属于业务调度，应写 next_retry_at 并发送延迟消息，然后 ACK 原消息。达到最大技术重试次数后进入 DLQ，但 DLQ 不是终点，需要隔离、分类、报警和带审计的选择性重放。

最关键的异常是供应商已经受理但本地超时。提交前我会先创建 attempt 和稳定幂等键，超时后进入 SUBMIT_UNKNOWN，先通过幂等键或查询接口对账，不能直接重提，否则会产生重复视频和重复成本。最后再用 Reconciler 扫描长时间不推进的任务、Outbox、媒体产物和计费流水，保证即使 MQ、消费者或第三方出现故障，系统仍能恢复到一致状态。

---

7.20 十分钟深入讲解提纲

第 1 分钟：业务背景与边界

• AI 视频是长耗时、高成本、强异步任务。
• PostgreSQL 管事实，RocketMQ 管传递，Redis 管准入，对象存储管媒体。
• RocketMQ 不天然保证业务 exactly-once。

第 2 分钟：整体异步架构

• API 事务写任务、额度和 Outbox。
• Relay 发布消息。
• Scheduler、Poll、Fetch、Media、Render、Billing、Notification 独立消费。
• Reconciler 作为最终兜底。

第 3 分钟：可靠生产

• 解释数据库与 MQ 双写裂缝。
• Outbox 的事务边界。
• Relay 使用租约、批量领取和稳定 event_id。
• Broker ACK 超时后为何允许重复发送。

第 4 分钟：可靠消费

• 至少一次语义。
• Inbox 去重唯一键。
• 先 DB commit 后 ACK。
• DB commit 成功但 ACK 丢失的恢复过程。

第 5 分钟：消息与 Topic 设计

• Topic 按业务域、消息类型、SLA 和资源隔离。
• Tag 用于同域子类型。
• Consumer Group 对应独立订阅者。
• event_id、task_id、trace_id、schema_version 的作用。
• 消息只传控制信息，不传媒体数据。

第 6 分钟：顺序和状态机

• 全局顺序为何不可取。
• task_id 局部顺序的适用范围。
• FIFO 的头阻塞风险。
• 版本号和前置状态条件更新是最终防线。

第 7 分钟：重试、延迟和 DLQ

• 技术消费重试与业务延迟重试分离。
• 429/503 使用退避和 jitter。
• 延迟消息到达后仍校验数据库状态。
• Poison Message 快速隔离。
• DLQ 的责任人、报警和选择性重放。

第 8 分钟：第三方提交不确定性

• 创建 attempt 和供应商幂等键。
• Provider 已受理、本地超时的时间线。
• SUBMIT_UNKNOWN、查询、对账和有限重试。
• 重复生成与重复计费风险。

第 9 分钟：容量与高可用

• λ_msg = λ_task × fanout。
• 消费能力和积压清空公式。
• 轮询消息放大、热点组、无界 goroutine 等瓶颈。
• MQ 故障时继续写 Outbox，长期故障则限流和暂停低优先级入口。

第 10 分钟：方案取舍

• Outbox 与事务消息比较。
• 本项目默认 Outbox 的原因。
• 无论选择哪种方案，消费幂等、外部副作用幂等和 Reconciler 都不能省略。
• 用一句话收尾：可靠异步的目标不是“消息永不重复”，而是“消息可以重复，业务结果不能错误”。

---

7.21 本章检查清单

• 业务写入与 Outbox 在同一 PostgreSQL 事务中。
• Outbox 重发保持同一 event_id。
• 每个核心 Consumer 有 inbox_dedup 或等价唯一约束。
• 所有 ACK 都晚于本地持久化。
• 外部 HTTP 调用不放在长数据库事务内。
• Provider Submit 超时进入不确定态，不盲目重提。
• 计费流水有业务唯一键。
• MQ 重试与业务延迟重试已分开。
• 延迟消息到达后会校验任务当前状态。
• FIFO 只用于局部必要顺序。
• 消息中不包含媒体二进制和不必要的敏感数据。
• DLQ 有报警、负责人、隔离表和审计重放工具。
• 监控 Outbox 年龄、消费 lag、最老消息年龄、DLQ 和 SUBMIT_UNKNOWN。
• MQ 长时间故障时有入口限流和服务降级。
• Reconciler 能从 PostgreSQL、供应商和对象存储恢复遗漏状态。

---

参考资料

1. Apache RocketMQ 官方文档：Consumption Retry。
2. Apache RocketMQ 官方文档：Sending Retry and Throttling Policy。
3. Apache RocketMQ 官方文档：Ordered Message。
4. Apache RocketMQ 官方文档：Delay Message。
5. Apache RocketMQ 官方文档：Transaction Message。
6. Apache RocketMQ 官方文档：Message Storage and Cleanup。
7. Apache RocketMQ 官方文档：Consumer Load Balancing。
8. Apache RocketMQ 官方文档：Metrics。
9. Apache RocketMQ 官方文档：Security。
10. Apache RocketMQ 官方文档：Go Client SDK。