消息队列协议

消息系统既定义字节怎样交换，也定义谁保存消息、何时确认以及消费者如何恢复。比较 RabbitMQ、Kafka 与 NATS 时，先分清标准协议、产品和产品内的存储模式；“智能 Broker、愚笨 Broker、最小 Broker”不足以解释真实的故障行为。

同一个产品内，存储类型、确认方式和复制配置的影响，往往大于产品名称之间的差异。RabbitMQ AMQP 1.0、NATS JetStream

先回答“成功”究竟确认了什么

生产者将字节交给 socket，不等于 Broker 已可靠接收；Broker 返回成功，不等于订阅者已经修改了业务数据库；消费者确认或提交 offset，也不自动证明本地事务已提交。这些边界发生在不同进程，必须有明确故障策略。

若消费端先确认再执行业务，崩溃可能漏处理；先执行业务再确认，确认丢失可能导致重复。把去重与本地业务效果放在一个事务内可以覆盖该局部窗口，外部 HTTP 副作用仍要靠外部幂等契约或补偿。具体接收模型见Webhook 与回调。

AMQP 0-9-1：路由到队列，再从队列交付

发布者把消息发给 exchange，binding 决定进入哪些队列。同一个 fanout exchange 绑定三个队列时，三个队列可各保留一份待消费状态；同一个队列上的三个竞争消费者，通常是在分摊该队列的交付，不是各收一份广播。

队列 durable、消息 persistent、复制和 publisher confirms 是不同配置。持久队列不意味着它所有消息天然按相同方式落盘；消费者 ACK 也不说明其他绑定队列中的副本已被消费。RabbitMQ Queues

发布确认与未路由消息

RabbitMQ publisher confirms 是对 AMQP 0-9-1 的扩展，用来确认发布侧处理。消息没有路由到任何队列时，也可能收到发布确认；需要发现这种情况时，还要正确使用 mandatory/return 等机制。发布确认与消费 ACK 完全独立。RabbitMQ confirms

消费 delivery_tag 的作用域是 channel，要在收到它的同一 channel 上确认；重连后不能把旧 tag 当作持久消息 ID。multiple=true 会确认到指定 tag 为止的多个未确认交付，若中间还有未完成工作，就会制造漏处理窗口。

手动确认模式下，连接或 channel 关闭通常使未确认交付重新排队；自动确认模式不提供同样保护。requeue 也可能改变观察到的顺序，反复 nack/requeue 会形成热循环，应记录尝试次数并对暂时失败与不可处理消息采用不同策略。

Prefetch 不是公平性开关

prefetch 限制未确认的交付数量，0 通常表示不设置这个数量上限。RabbitMQ 对非 global prefetch 常按每位消费者应用；两个消费者各 prefetch=10，不应把总上限误当成整个 channel 只有 10。Consumer Prefetch

设置为 1 会限制每位消费者的在途消息，能减少慢消费者提前占住一大批工作，但吞吐还受确认往返与处理时间影响，也不保证各消费者处理工作量或耗时完全公平。设置很大则可能造成客户端内存增长、重投风暴和尾部等待。

字节帧与业务消息也要分层

AMQP 0-9-1 帧包含类型1 B、channel2 B、payload 长度4 B、payload，以及末尾0xce。一次发布可以由方法帧、内容头帧与多个内容体帧组成，不能按一次 socket read 或一个帧当作一条完整业务消息。心跳是类型8、channel0、空 payload 的8 B帧。RabbitMQ 帧格式勘误

import struct
heartbeat = struct.pack('!BHI', 8, 0, 0) + b'\xce'
assert heartbeat.hex() == '08000000000000ce'
assert len(heartbeat) == 8
print('AMQP 0-9-1 空心跳帧长度通过')

AMQP 1.0：连接、会话与链路的流控

AMQP 1.0 不只是 0-9-1 换一个版本号。端点在 connection 内建立 session，再 attach 单向 link；消息由 transfer 承载，通过 delivery state 与 settlement 表达结果，link credit 约束发送权限。它不要求所有实现都采用完全相同的 exchange/queue 管理接口。

同样叫 AMQP 的两个客户端如果版本不同，不能直接互通。迁移时应同时检查协议版本、地址映射、认证、确认模式和产品扩展；用“双方都支持 AMQP”不足以证明行为一致。RabbitMQ AMQP 1.0 概念

Kafka：日志位置、组进度与保留是不同状态

Kafka 把记录追加到 topic 的 partition 中，每条记录有该分区的 offset。常规 consumer group 消费时，各成员分配分区并拉取记录；不同组可以独立读取同一日志。组的提交位置通常由 Kafka 协调保存，不是只存在消费进程内。

提交 offset 不删除对应记录。日志按时间、容量或 compaction 等策略保留，消费者落后到数据已清理后，原位置就可能无法重放。所谓“消费者自己跟踪 offset”不能进一步推导 Broker 不管理组、事务或恢复状态。Kafka Design

应用一次拉取 100–103 并发处理，若 100 和 102 先完成，安全恢复位置仍是 101。只有 101 也完成，才能推进到 103；如果先提交 103，崩溃后会跳过未完成的 101。

def resume_offset(first, completed):
    result = first
    while result in completed:
        result += 1
    return result

assert resume_offset(100, {100, 102}) == 101
assert resume_offset(100, {100, 101, 102}) == 103
assert resume_offset(100, {101, 102}) == 100
# 明确假设仅有这一段连续 offset；压缩日志中实际返回位置可有空洞。
print('连续完成前缀与恢复位置通过')

这个模型的连续整数用于说明提交规则；真实日志可能有事务、压缩或其他原因造成 offset 空洞，不能因为没收到某个整数就无限等待。应依据实际返回记录与分区进度维护已处理前缀。

非连续 offset、批次边界以及再平衡后的旧任务处理，继续看分布式系列的 消费组与协作。本篇保留协议层面的最小例子，完整的恢复进度管理在那里展开。

Kafka wire protocol 不是固定调用流水线

请求通常带长度前缀、API key/version、correlation ID 与相应版本头部；响应通过 correlation ID 关联请求。ApiVersions 用于能力协商，Metadata 提供节点与分区信息，Produce/Fetch 处理记录，组协议与 OffsetCommit 管理消费协调。生产者不会为了发布每条记录先执行一次 Fetch。

请求/响应 schema 随 API 版本变化，较新 flexible versions 还有紧凑编码和 tagged fields；不能用一个固定 struct 解析所有版本。记录还可能批处理与压缩，一次请求不等于一条消息。Kafka Protocol

acks=all 的确认要结合 ISR 与最小同步副本等配置理解，不代表每个配置副本都永远已经同步。幂等生产者管理相应生产会话与分区序列；应用随意重新创建生产者再提交同一业务事件，仍需考虑业务去重。Kafka 事务可以协调 Kafka 输出和消费位置，不把外部数据库自动纳入事务。

102处理完了，为什么还只能提交101？

先完成100和102，安全恢复位置停在101；若误提交103，崩溃后会跳过101。补完101时可以一次推进到103。与TCP累计确认相似的是连续前缀约束，不同的是这里的“完成”由业务消费定义。

Core NATS：文本命令与按字节定界的 payload

Core NATS 通过 subject 进行在线发布订阅，queue group 用来在组内分摊交付。它不默认保存离线订阅者漏掉的历史消息；PING/PONG 也不意味着所有订阅者已经处理完业务。

客户端协议命令以文本表示，但 payload 可以是二进制，长度字段是字节数而不是字符数。SUB 指定 subject、可选 queue group 与 sid；PUB 指定 subject、可选 reply subject 与 payload size。sid 是该连接上的订阅标识，不是消息 ID。NATS Client Protocol

payload = '温度=23'.encode('utf-8')
assert len(payload) == 9
packet = b'PUB sensors.temp ' + str(len(payload)).encode('ascii') + b'\r\n' + payload + b'\r\n'
line, rest = packet.split(b'\r\n', 1)
size = int(line.split()[-1])
assert rest[:size].decode('utf-8') == '温度=23'
assert rest[size:] == b'\r\n'
print('NATS UTF-8 payload 字节定界通过')

NATS 过滤器 * 匹配一个 token，末尾 > 匹配一个或多个 token；不要把它与 MQTT # 的零层匹配直接互换。request/reply 利用 reply subject 返回消息，但请求超时仍可能发生在远端已经执行之后。

JetStream：持久流与有状态消费者

JetStream stream 捕获指定 subjects 的消息，可配置内存/文件存储、复制和保留策略；consumer 保存交付与确认状态。它支持 pull 等消费方式，不能把所有 NATS 功能都归为无状态、仅推送或必定最低延迟。JetStream

发布方通过 JetStream publish acknowledgment 确认进入 stream；消费方 ACK 推进 consumer 状态。未确认消息可按配置重投，AckWait 与最大尝试次数需要与处理时间配合。不同 retention policy 对确认后是否删除消息的影响不同，不能统一解释为“ACK 只是 offset”或“ACK 永远删除”。

消息去重窗口和双重确认等功能可以减少协议层重复，但需要对应配置与 API。窗口外重发、应用新建事件或外部副作用，都不能仅靠一句 exactly-once 消除。JetStream 消费行为

用积压与恢复要求作比较

假设输入每秒 1200 条、消费者稳定处理 1000 条，每秒积压 200 条；十分钟后积压 120,000 条。若消息加元数据平均 1 KiB，约 117.2 MiB，还不含复制副本。换成更轻量的 wire protocol 并不会消除生产和消费之间的长期差额。

性能报告要同时列出包长、批次、压缩、确认、复制、磁盘与消费者工作量。没有这些条件的“每秒百万条、固定 0.1 ms”无法作为选型证据，协议复杂度也不能直接映射成端到端延迟。