messaging-streaming

消息与流式处理

将工作移出同步请求路径，让生产者与消费者解耦，这样缓慢、突发或容易失败的操作就不会阻塞调用方。这里很容易出错：队列会悄然改变投递和顺序保证；在负载下，它既可以平稳吸收流量峰值，也可能成为掩盖系统崩溃的因素，直到积压无法挽回。

何时使用

某个步骤太慢，无法内联执行（图像转码、扇出、调用第三方服务）；写入路径流量突发，需要缓冲来平滑峰值（→ back-of-the-envelope，了解峰值系数）；两个服务必须解耦，以便其中一个可以独立故障或部署；或者许多消费者需要读取相同的事件流。异步交接能带来更快的响应、更强的隔离性和弹性（分别扩展生产者和消费者）。

何时不使用

调用方需要立即获得结果才能继续（同步读取、确认前的余额检查）——队列只会增加延迟，并增加工作丢失的可能。单个请求内需要强读后写一致性。简单的进程内工作，用函数调用就能完成。不要在问题尚未证明有必要时就加入消息代理（YAGNI）：这是一个新的有状态系统，需要运维、监控，并在故障时进行分析。“我们迟早会需要 Kafka”只是堆砌名称，不是需求。

先明确

• 同步还是异步？ 调用方需要内联获取结果，还是可以接受即发即走？这决定了这里是否需要队列。
• 所需的投递保证——允许消息丢失吗（至多一次），还是必须处理每条消息（至少一次 + 幂等消费者）？
• 顺序——消息必须按顺序处理吗？是全局有序，还是按键有序（按用户、按账户）？全局有序成本高；通常按键有序就够了。
• 吞吐量和保留时间——峰值时每秒有多少条消息？需要保留多长时间以便重放？（→ back-of-the-envelope。）是一次性任务还是可重放的日志？
• 消费者数量和模式——一个工作池负责取出并处理任务，还是多个独立订阅者各自读取每个事件？
• 故障处理——持续失败的消息会如何处理？会被放到哪里？由谁查看？

选项

同步还是异步——选择工具前先确定这一点。 如果调用方需要结果才能继续，且调用快速可靠，就采用同步方式；直接请求更容易构建、跟踪和分析。如果工作耗时较长、负载突增、需要扇出，或者调用方可以稍后再处理结果，就采用异步方式——这会牺牲即时一致性和简单的堆栈跟踪，换取响应能力和隔离性。只有选择异步后，下面的选项才适用。通过队列构建请求/回复来伪造同步应答是一种反模式——直接调用才是更好的设计。

队列（工作/任务队列）——一个逻辑消费者组竞争处理消息；一条消息只会投递给一个工作进程，并在确认后移除。适用于有后台任务或命令需要近似恰好处理一次，且工作进程应能扩展以清空积压的场景。

发布/订阅（扇出）——每个订阅者都会收到每条消息的副本；生产者不知道订阅者是谁。适用于多个独立消费者需要对同一事件作出反应（通知、索引、审计），且松耦合很重要的场景。

流（持久、可重放的日志）——仅追加、分区且保留的日志；消费者自行跟踪偏移量，并可重放历史记录。适用于设计需要按分区排序、多个消费者处于不同位置、事件溯源或重新处理的场景（事件溯源/Outbox 参见 data-storage）。

持久化工作流（编排引擎）——进程崩溃后仍能继续运行的代码；引擎会持久化每个步骤，并从中断处恢复，内置重试、计时器和补偿机制。适用于多步骤流程包含重试、人工等待和回滚（Saga），否则就会变成脆弱的手工队列、状态标记和 cron 任务网状结构的场景。

交付语义适用于以上所有方式：至多一次（发送后不管，消息可能丢失）、至少一次（持续重试直到确认，可能重复——实际中的默认选择）、恰好一次（不丢失、不重复）。真正端到端的恰好一次难以实现：消息代理的“EOS”（例如 Kafka）仅限集群内部，因此跨系统时总是要通过至少一次 + 幂等/去重消费者来实现（api-design 中的幂等键）。具体机制请参阅 references/deep-dive.md。

权衡

选项	解决的问题	代价	何时更换
队列（工作队列）	解耦并缓冲；扩展工作进程以清空积压	至少一次意味着可能重复；无法保证跨工作进程的顺序	需要重放或多个独立消费者 → 流/发布订阅
发布/订阅（扇出）	一个事件触发 N 个解耦的响应；可自由添加消费者	无法重放（临时）；缓慢的订阅者可能落后或丢失消息；扇出会放大负载	需要历史记录/重放或按键排序 → 流
流（日志）	按分区排序、重放、多消费者、事件溯源	运维负担更重；分区键有造成热点分片的风险；消费者必须管理偏移量	简单的一次性任务不需要日志 → 队列
持久化工作流	崩溃安全的长时运行 Saga；内置重试/补偿机制	引入新的运行时和编程模型；增加延迟开销；受限于引擎语义	单个无需编排的异步步骤 → 普通队列
至少一次交付	重试/崩溃时不会丢失消息	可能重复——消费者必须具备幂等性（api-design 中的幂等键）	可以接受丢失，且去重成本不值得 → 至多一次
恰好一次（有效实现）	不丢失，且副作用不重复	成本/复杂度高；适用范围通常较窄（限于单个消息代理内，而非跨系统）	幂等的至少一次已足够（通常确实如此）

压力下的行为

Broker 的全部职责就是吸收流量峰值，但它也可能 掩盖 故障。

• 积压增长 / 无界队列：生产者的速度超过消费者；队列增长到超出内存容量，溢出到磁盘并进一步变慢。端到端延迟悄然攀升，吞吐量看起来却正常。*缓解措施：*采用背压——限制队列大小，队列满时让生产者降速或返回 503（并采用退避），而不是无限缓冲。监控队列深度和消息年龄，而不只是速率。
• 重试风暴 / 毒消息：始终处理失败的消息会被无限重新投递（至少一次），耗尽消费者容量，并在下游已不堪重负时进一步放大负载。*缓解措施：*限制重试次数，并采用退避 + 抖动，然后将消息路由到死信队列（DLQ），避免毒消息阻塞处理流程，并供人工检查。（重试、退避、通过 DLQ 隔离问题以及背压由 resilience-failure 负责。）
• **重复放大：**在重试过程中，同一副作用（扣款、发送邮件）可能触发两次，除非消费者进行去重。*缓解措施：*幂等键 / 去重表。
• **热点分区：**在流中，倾斜的分区键（某个名人、某个租户）会使单个分区过载，而其他分区处于空闲状态——这与 data-storage 分片中的热点键问题相同。*缓解措施：*换用更合适的键、进一步划分分区，或进行批处理。
• **扇出中的慢速消费者：**一个落后的订阅者可能导致阻塞或丢弃消息；隔离消费者，避免一个消费者拖慢其他消费者。

**监控：**队列深度、最旧消息年龄 / 消费者滞后（最重要的单一指标）、重新投递 / DLQ 速率、消费者吞吐量与生产者速率，以及端到端延迟。

如何应用

1. **明确输入。**先确定采用同步还是异步；如果调用方需要直接获取结果，就此打住——不要用 broker。然后确定投递保证、顺序范围、吞吐量 / 保留时间、消费者模式和故障处理方式（参见 先明确需求）。
2. **根据权衡表选择形态。**一个工作池负责处理任务 → 队列；对一个事件作出多种独立响应 → 发布 / 订阅；需要顺序处理、重放或多个偏移量 → 流；多步骤重试 / 补偿 Saga → 持久化工作流。选择满足约束的最简单方案。
3. **设置关键参数。**事先确定投递语义（默认采用至少一次 + 幂等消费者）、确认时机（处理后还是处理前）、分区 / 顺序键、保留窗口，以及进入 DLQ 前的重试上限。
4. **对方案进行压力测试。**逐一检查积压增长、重试风暴 / 毒消息、重复放大、热点分区和扇出中的慢速消费者。在适用的地方添加背压、DLQ 和去重。
5. **用数字估算容量。**计算峰值生产速率与持续消费速率、积压清空速率 consume_rate − produce_rate、分区数与消费者并行度，以及存储量 = 速率 × 消息大小 × 保留时间（→ back-of-the-envelope）。
6. **选择提供商。**默认采用通用方案；只有当用户指定了云平台，且托管服务的限制会改变选择时，才查看对应的提供商文件。
7. 在回答中展示权衡——提供一个 决策 | 解决的问题 | 加剧的问题 | 在何时更改… 表格，每个此处作出的选择各占一行（例如所选方案、存储方式、故障策略）；最后一列写出具体的触发条件（某个数字或可观测信号）。然后指出最先出现故障的部分。

应做与不应做

应做

• 在命名任何消息代理之前，先确定采用同步还是异步；保持同步工作的同步性。
• 默认采用至少一次投递，并确保消费者具备幂等性（→ api-design 密钥）。
• 限制队列容量并施加背压；针对最早消息的等待时间 / 消费者滞后设置告警。
• 通过退避 + 抖动限制重试次数，然后将有毒消息路由到 DLQ。
• 预先定义消息封套（message_id、架构版本、键、trace_id）。

不应做

• 不要在没有具体数字或耦合问题作为依据时，就为了“以后扩展”添加消息代理。
• 不要通过队列构建请求 / 响应模式，来伪造同步答复。
• 如果幂等的至少一次投递已经足够，就不要追求恰好一次。
• 不要缓冲无上限的积压消息——这会掩盖故障，直到问题无法挽回。
• 如果按键排序就够了，不要选择全局排序保证。

重要数据

选择之前先量化：生产峰值速率与持续消费速率（如果生产者可能长时间超过消费者，必须施加背压并设置队列深度告警）、保留窗口（决定存储量 = 速率 × 消息大小 × 保留时长），以及分区数量（限制消费者并行度——每个组每个分区对应一个消费者）。积压消息会以 (consume_rate − produce_rate) 的速率消退；如果该值为负，积压就永远不会消退。使用 back-of-the-envelope 估算峰值系数、消息大小和存储量；不要在这里重复其中的表格。

接口示意

消息是一份契约。应明确地定义消息，而不是笼统地称为“事件”：

• **封套：**稳定的 message_id（用于去重）、type / schema_version、partition_key / 排序键、timestamp，以及用于关联的 trace_id。
• **负载：**采用版本化架构。扇出到多个订阅者时，优先使用事件事实（OrderPlaced{order_id, total}），而不是命令；保持精简并向前兼容。
• **确认契约：**消费者在什么时候确认消息——副作用发生之前还是之后？处理后确认可实现至少一次；处理前确认可实现至多一次。
• **DLQ 格式：**失败的消息应保留原始封套，以及失败原因和尝试次数，以便检查和重放。

选择提供商

默认采用上述通用方案。如果用户指定了云服务，请阅读 references/providers/<provider>.md，了解托管服务映射、配额 / 限制以及提供商特有的权衡。如果该提供商没有对应文件，则采用通用方案。

图示

若要展示生产者 → 消息代理 → 消费者的路径、扇出到多个订阅者的过程，或重试 → DLQ 流程，请使用插件内的 architecture-diagram 技能。快速内联示意：producer → [queue] → workers ─fail→ [DLQ]；主路径用实线，DLQ 分支用虚线。

相关构件

• resilience-failure — *相关概念归属：*重试 / 退避 / 抖动、死信队列，以及作为故障遏制手段的背压。本技能介绍这些概念；该技能负责调整它们。
• api-design — 依赖于其幂等键；该机制使至少一次投递变得安全。
• data-storage — 与本技能配合用于事件溯源和事务性 outbox；*相关概念归属：*流式处理继承的热分片 / 分区键问题。
• consistency-coordination — *相关概念归属：*排序保证、恰好一次与幂等性之间的取舍，以及持久化工作流背后的 Saga / 分布式事务理论。
• back-of-the-envelope — *为容量估算提供依据：*用于确定消息代理规模的生产 / 消费速率、保留时长和存储量。
• system-design — 负责编排并在工作转为异步时将流程路由至此。

参考资料

• references/deep-dive.md — 投递保证机制（确认、可见性超时、偏移量、幂等 / 去重消费者、事务性 outbox）、排序内部机制、分区、DLQ 设计，以及持久化工作流引擎优于手动构建队列 + 重试 + Saga 的情形。设计消息传递层的详细方案时请阅读。
• references/providers/{generic,aws,azure,gcp,temporal}.md — 各环境下的服务映射、会影响决策的限制以及注意事项。