43.1 消息模型 (点对点 / 发布订阅)
一句话
点对点是“一条消息只交给一个消费者”（队列语义），发布订阅是“一条消息给每个订阅者各一份”（广播语义）；选型的分水岭是这条消息该被处理一次，还是被所有关心它的人各看一次——成本、重复与顺序问题都由这个选择决定。
两种模型，一张图
点对点（Queue / 竞争消费） 发布订阅（Topic / 广播）
producer ──► [ 队列 ] ──► consumer A producer ──► [ Topic ] ──┬──► subscriber 1（各一份）
└► consumer B ├──► subscriber 2（各一份）
└► consumer C └──► subscriber 3（各一份）
| 维度 | 点对点 | 发布订阅 |
|---|---|---|
| 一条消息投几次 | 1 次 | 订阅者数量次 |
| 消费语义 | 竞争消费（load balance） | 广播（每个订阅者都收到全量） |
| 扩消费者的效果 | 提高吞吐（分摊） | 每个新订阅者都增加一份投递与存储 |
| 典型场景 | 任务队列、削峰、订单处理 | 事件通知、缓存失效、多下游数据同步 |
| 消息保留 | 消费后删除（或按时间） | 按订阅位点/保留期，各订阅者独立进度 |
Kafka 同时支持两种：同一个 Topic 内，不同消费组（consumer group）之间是发布订阅，组内的消费者之间是点对点。这不是“两种产品”，而是用消费组这个维度把两种语义组合起来——这是它比传统队列灵活的地方。
数字一：放大倍数与存储
100 万条消息、3 个消费者、5 个订阅者，每条 1 KiB：
| 模型 | 投递次数 | 存储（7 天） |
|---|---|---|
| 点对点 | 100 万 | 1.0 GB/天 → 7 GB |
| 发布订阅 | 500 万（5 倍） | 5.1 GB/天 → 36 GB |
🔑 发布订阅的成本是订阅者数量 × 消息量。加一个订阅者不是“多一个消费者”，而是“多一份全量流量与存储”。所以“把所有事件都广播出去”很容易在半年后变成账单事故。
如果下游只是“关心其中一部分消息”，正确做法不是多一个订阅者，而是：
| 做法 | 效果 |
|---|---|
| 按 topic 拆分（订单/支付/库存各一个） | 每个订阅者只收自己关心的 |
| 服务端过滤（按 tag/header） | 网络与存储只传匹配的 |
| 消费后转发（一个订阅者扇出到多个内部模块） | 跨订阅者的扇出只发生一次 |
数字二：消费能力不足会怎样
生产 5000 条/秒，3 个消费者各 1000 条/秒：
消费能力 = 3 × 1000 = 3000/s
积压速度 = 5000 − 3000 = 2000/s
10 万条队列：50 秒填满
填满之后只有三条路：丢消息（最坏）、生产端阻塞/背压（保护系统但拖慢上游）、扩容消费者（受分区数限制）。这就是 43.5 积压与背压（43.5）的核心：队列不是无限缓冲，它是“有界的延迟”。
| 现象 | 含义 | 动作 |
|---|---|---|
| 消费延迟稳定在几百 ms | 正常波动 | 观察 |
| 延迟持续线性上涨 | 消费能力 < 生产 | 加消费者/加分区/优化消费逻辑 |
| 队列长度周期性归零又涨 | 批次/定时任务 | 削峰或调批大小 |
| 重平衡频繁 | 消费者抖动 | 调心跳与超时（见 43.4 消费组与再均衡） |
数字三：至少一次语义下的重复
消费者每 100 条 ack 一次，处理到第 500,050 条时宕机——本批已处理 50 条但还没 ack，重启后这 50 条会被重投：
| 指标 | 数值 |
|---|---|
| 投递次数 | 1,000,050 |
| 处理的唯一消息 | 1,000,000 |
| 重复 | 50（0.005%） |
| 重复的绝对量（1 亿条/天） | 5,000 条/天 |
0.005% 看起来很小，但在 1 亿条/天的规模下就是每天 5000 条重复。“精确一次”不是 broker 的承诺，而是消费端用去重键换来的（见 25.4 幂等设计与去重）。broker 能保证的是：
| 语义 | broker 保证 | 代价 |
|---|---|---|
| at-most-once | 最多一次，可能丢 | 先 ack 再处理，最快 |
| at-least-once | 至少一次，可能重复 | 处理完再 ack，最常用 |
| exactly-once（broker 内） | 事务 + 幂等生产 + 位点原子提交 | 性能下降、跨系统不成立 |
跨系统（MQ → 数据库、MQ → 另一个 MQ）时，“精确一次”必须由业务侧的去重/事务来保证。
拉还是推
| 模式 | 做法 | 优点 | 缺点 |
|---|---|---|---|
| 推（push） | broker 主动发给消费者 | 延迟低 | 消费者能力弱时被打爆，需要背压 |
| 拉（pull） | 消费者按需拉一批 | 消费者控制节奏，天然背压 | 空轮询浪费，需要长轮询/批量 |
| 长轮询 | 拉不到就挂起，有数据立刻返回 | 兼顾延迟与背压 | 连接数与超时管理 |
Kafka 是拉模型（靠 fetch.min.bytes + 长轮询降延迟），RabbitMQ 推模型（靠 prefetch limit 做背压）。选择的关键不是“谁更快”，而是谁来决定消费速率：消费者自己决定（拉）还是 broker 决定（推）。
顺序：只在分区内成立
同一分区（partition）内：严格有序
跨分区：无序
要保序，必须让“需要保序的消息”落在同一分区：用业务键（订单 ID、用户 ID）做分区键。代价是热点：大客户的订单全落在同一分区，吞吐受单分区上限约束（见 43.3 顺序与分区）。
| 需求 | 做法 |
|---|---|
| 同一订单的事件有序 | 用订单 ID 做分区键 |
| 全局严格有序 | 单分区（吞吐上限 = 单消费者能力） |
| 只需“同一实体有序” | 实体 ID 做 key，接受跨实体乱序 |
| 乱序也能处理 | 用版本号/时间戳做幂等与覆盖 |
可运行的投递账本
package main
import "fmt"
func main() {
const messages = 1_000_000
const subscribers = 5
const consumers = 3
// ① 点对点 vs 发布订阅：投递次数
p2p := messages
pubsub := messages * subscribers
fmt.Printf("① %d 条消息，%d 个消费者，%d 个订阅者\n", messages, consumers, subscribers)
fmt.Printf(" 点对点：%d 次投递（每条只给一个消费者）\n", p2p)
fmt.Printf(" 发布订阅：%d 次投递（每个订阅者各一份，放大 %.0f 倍）\n", pubsub, float64(pubsub)/float64(p2p))
// ② 存储：1 KiB/条，保留 7 天
const kib = 1024
fmt.Printf("② 存储（1 KiB/条）：点对点 %.1f GB/天，发布订阅 %.1f GB/天（7 天：%.0f vs %.0f GB）\n",
float64(messages*kib)/1e9, float64(pubsub*kib)/1e9,
float64(messages*kib*7)/1e9, float64(pubsub*kib*7)/1e9)
// ③ 消费能力与积压：生产 5000/s，每消费者 1000/s
produce, perConsumer := 5000.0, 1000.0
capacity := perConsumer * consumers
backlogPerSec := produce - capacity
fmt.Printf("③ 生产 %.0f/s，%d 个消费者 × %.0f/s = %.0f/s → 每秒积压 %.0f 条\n",
produce, consumers, perConsumer, capacity, backlogPerSec)
if backlogPerSec > 0 {
limit := 100_000.0
fmt.Printf(" 10 万条队列 %.0f 秒填满；之后要么丢消息，要么生产端阻塞（背压）\n", limit/backlogPerSec)
}
// ④ at-least-once：消费者每 100 条 ack 一次，崩溃时未 ack 的那一批要重投
const batch = 100
crashAt := 500_050 // 处理到第 500,050 条时宕机：本批已处理 50 条，还没到 100 条一次的 ack
deliveries := messages
duplicates := crashAt % batch
deliveries += duplicates
fmt.Printf("④ 至少一次语义：%d 次投递处理 %d 条消息，崩溃重投 %d 条（重复率 %.3f%%）\n",
deliveries, messages, duplicates, float64(duplicates)/float64(messages)*100)
fmt.Println(" 精确一次（幂等消费）不是 broker 的承诺，是消费端用去重键换来的")
}
输出：
① 1000000 条消息，3 个消费者，5 个订阅者
点对点：1000000 次投递（每条只给一个消费者）
发布订阅：5000000 次投递（每个订阅者各一份，放大 5 倍）
② 存储（1 KiB/条）：点对点 1.0 GB/天，发布订阅 5.1 GB/天（7 天：7 vs 36 GB）
③ 生产 5000/s，3 个消费者 × 1000/s = 3000/s → 每秒积压 2000 条
10 万条队列 50 秒填满；之后要么丢消息，要么生产端阻塞（背压）
④ 至少一次语义：1000050 次投递处理 1000000 条消息，崩溃重投 50 条（重复率 0.005%）
精确一次（幂等消费）不是 broker 的承诺，是消费端用去重键换来的
参数怎么给
| 你的场景 | 模型 | 关键参数 |
|---|---|---|
| 任务队列/削峰 | 点对点 | 分区数 ≥ 消费者数；可见性超时 > 处理时间 |
| 事件广播（缓存失效、配置变更） | 发布订阅 | 订阅者数量受控；消息带版本号，消费端幂等 |
| 多下游数据同步 | 发布订阅 + 服务端过滤 | 按 tag 过滤，避免全量复制 |
| 同一实体的状态变更 | 点对点 + 分区键 | 分区键 = 实体 ID，保序 |
| 跨系统通知 | 点对点 + 重试 + 死信 | 至少一次 + 幂等消费 + 死信队列 |
| 大促削峰 | 点对点 + 有界队列 | 队列长度上限 + 生产端背压 |
把语义落到 Kafka 参数：producer / consumer 两份
下面两份用 Kafka 4.3 单节点跑通。注意 4.x 的脚本去掉了
.sh后缀，控制台工具改用--command-config（旧的--producer.config/--producer-property已弃用）。
# producer.properties
bootstrap.servers=10.0.1.21:9092,10.0.1.22:9092,10.0.1.23:9092
acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5
retries=2147483647
delivery.timeout.ms=120000
linger.ms=5
batch.size=32768
compression.type=lz4
# consumer.properties
bootstrap.servers=10.0.1.21:9092,10.0.1.22:9092,10.0.1.23:9092
group.id=order-consumer
enable.auto.commit=false
isolation.level=read_committed
max.poll.records=500
| 参数 | 值 | 对应哪个语义 | 配错的后果 |
|---|---|---|---|
acks=all | 所有 ISR 确认 | 至少一次、不丢 | acks=1：leader 挂了就丢；acks=0：发出去就不管 |
enable.idempotence=true | 开 | 生产端重试不写重复 | 重试产生重复，消费端被迫全量幂等 |
max.in.flight.requests.per.connection | 5 | 幂等下仍保序的上限 | 调大 + 非幂等 = 重试乱序 |
delivery.timeout.ms | 120s | 重试总预算，要大于 broker 的请求超时 | 太小：高峰期大量失败；太大：故障时长时间挂住 |
linger.ms + batch.size | 5ms / 32KB | 吞吐与延迟的折中 | 0ms 小批：吞吐低；太大：端到端延迟上升 |
compression.type | lz4 | 带宽与 CPU 的折中 | 不压缩：跨 AZ 流量成本高 |
enable.auto.commit | false | 至少一次（处理完再提交） | 自动提交 = 先提交后处理，崩了就丢 |
isolation.level | read_committed | 事务消息不读未提交 | read_uncommitted 会读到回滚的事务 |
max.poll.records | 500 | 单次处理量 × 耗时 < max.poll.interval.ms | 处理超时触发 rebalance，重复消费 |
broker / 主题级还要配两条，否则上面的 acks=all 会被架空：min.insync.replicas=2（三副本集群才有意义，单节点验证不了）、unclean.leader.election.enable=false（宁可不可用，也不让落后副本当 leader）。acks=all 只在「ISR 数量 ≥ min.insync.replicas」时才真的等多数派。
# 主题的 ISR 与 min.insync
kafka-topics --bootstrap-server 10.0.1.21:9092 --describe --topic orders
# 生产 + 消费各跑一次（配置随 --command-config 传入）
printf 'k-1\n' | kafka-console-producer --bootstrap-server 10.0.1.21:9092 \
--command-config producer.properties --topic orders
kafka-console-consumer --bootstrap-server 10.0.1.21:9092 \
--command-config consumer.properties --group order-consumer --topic orders \
--from-beginning --max-messages 1
# 积压：LAG 持续上涨 = 消费能力不足，不是 broker 问题
kafka-consumer-groups --bootstrap-server 10.0.1.21:9092 --describe --group order-consumer
⚠️ 不要指望
enable.idempotence给你端到端“恰好一次”：它只覆盖「生产端重试」这一小段，限单分区、单生产者会话。消费端重启、跨系统写库、跨 Topic 转发都会重新引入重复——幂等仍要业务自己做。
反模式
| 反模式 | 后果 |
|---|---|
| 给每个下游都开一个订阅者收全量 | 流量与存储按订阅者数量线性放大 |
| 把点对点当广播用（多个 consumer 各收一份） | 同一任务被重复执行 |
| 把广播当点对点用（只起一个消费者） | 订阅者之间抢消息，各自只看到一部分 |
| 用消息队列当数据库查历史 | 保留期一过就查不到；查询能力极弱 |
| 消费能力不足却只调大队列 | 积压变成延迟，最终丢消息 |
| 跨分区还假设全局有序 | 乱序导致状态错乱 |
| 依赖 broker 的 exactly-once 跨系统 | 数据库与 MQ 不在一个事务里，仍会重复 |
| 每条消息单独 ack | ack 开销吃掉吞吐，批量 ack 更常见 |
| 没有死信队列 | 毒消息无限重投，堵住整个分区 |
自检
Q1 100 万条消息、5 个订阅者，发布订阅的投递次数是 500 万。如果这 5 个订阅者里有 3 个其实只关心 10% 的消息，成本能怎么降？
▸ 答案：不要用“5 个订阅者收全量”，改成服务端过滤或拆分 topic：① 按业务拆 topic（订单/支付/库存），每个订阅者只订阅自己关心的；② 用 tag/header 在 broker 侧过滤，只投递匹配的消息，网络与存储成本按 10% 计；③ 只保留 1 个订阅者做“扇出适配器”，它在内部把消息分发给多个模块——跨订阅者的复制只发生一次。三种手段都在降低“每人一份全量”的放大倍数。
Q2 生产 5000/s、3 个消费者各 1000/s，队列 10 万条。50 秒后会发生什么？为什么“把队列调大”不是解决方案？
▸ 答案：50 秒后队列满，之后只有三种结果：丢弃（丢消息）、阻塞生产端（背压，上游变慢）、或消费端扩容。把队列调大只是把“50 秒”变成“500 秒”，积压仍在增长，最终仍会满——而且更晚发现、故障更大。根因是消费能力 3000 < 生产 5000，要么加消费者/分区，要么优化消费逻辑（批量、异步、减少下游调用），要么在入口限流。队列是缓冲，不是产能。
Q3 0.005% 的重复率看起来可以忽略，为什么还必须做幂等？
▸ 答案：因为重复的绝对量随规模放大：1 亿条/天 × 0.005% = 5000 条/天；如果这些是支付、发券、扣库存，每天就有 5000 次资损或错账。而且重复不是均匀分布的——崩溃、重平衡、网络抖动会让重复集中在某一批消息上（比如一次重平衡重投几万条）。幂等的成本（一个去重键 + 唯一约束）远低于对账与赔付的成本。
Q4 为什么 Kafka 能在同一个 Topic 上同时提供“点对点”和“发布订阅”？
▸ 答案：因为它把“消费进度”放在消费组这一层：同组内的消费者竞争同一批分区（点对点，扩容即分摊）；不同消费组各自维护位点、各收全量（发布订阅）。所以两种语义不是二选一的产品形态，而是“组的数量”决定的配置。代价是：每个消费组都要独立保存位点、独立承受全量流量——订阅者越多，broker 的出口带宽与存储压力越大。
Q5 拉模型和推模型，哪个更容易做背压？为什么？
▸ 答案：拉模型天然有背压——消费者拉多少由自己决定，处理不过来就不拉（或降低 max.poll.records），broker 不会硬塞。推模型里 broker 掌握节奏，必须靠 prefetch limit、窗口、消费者主动 nack/pause 来限速，配置不当就会把消费者打爆（内存里堆满未处理消息）。所以高吞吐、消费者异构的场景多用拉（Kafka），低延迟、消费者能力可预期的场景多用推（RabbitMQ）。
Q6 acks=all + enable.idempotence=true，消息就不会重复了吗？
▸ 答案：只保证“生产端重试不会在 broker 写出重复”，且限单分区、单生产者会话。消费端重启后的重复投递、跨 Topic 转发、写库与发消息不在一个事务里，都会重新引入重复——所以消费端必须幂等（去重表/唯一键/版本号）。另外 acks=all 要配 min.insync.replicas≥2 才真的等多数派，否则等于只等了一个副本。