消息"丢了"或"重复了"，翻遍业务代码通常找不到原因——因为丢与重都不发生在应用层，发生在 broker 的副本确认逻辑和生产者重试逻辑之间那层空隙里。
三个配置项决定了这个空隙有多宽：acks 决定等几个副本写成功，min.insync.replicas 决定最少要几个副本活着，max.message.bytes 决定单条消息能不能进得去。它们不是独立开关，是一组互相牵制的约束。
1. 三种交付语义与 acks 的对应
| 语义 | 生产者配置 | 代价 | 真实保证 |
|---|---|---|---|
| 至多一次 | acks=0 |
最快 | 发出去就不管，网络抖动即丢 |
| 至少一次 | acks=1 或 all + 重试 |
中 | 会重复，需消费者幂等 |
| 精确一次 | acks=all + 幂等 + 事务 |
最慢 | 仅限 Kafka 内部链路 |
Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092,kafka-2:9092");
p.put(ProducerConfig.ACKS_CONFIG, "all");
p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
p.put(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE));
p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
p.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "120000");
acks=1 的语义常被误解成"写成功了"。它只表示 leader 写入本地日志成功，副本还没同步。leader 在同步完成前宕机、且该副本被选为新 leader，这条消息就凭空消失——日志里不会有任何错误。
开启 enable.idempotence=true 后，acks 会被强制提升到 all，retries 被强制设为极大值，max.in-flight 被限制在 5 以内。这三个不是建议，是幂等性的前置条件，配错会直接抛 ConfigException。
2. ISR 与 min.insync.replicas：可用性与持久性的交换
ISR（In-Sync Replicas）是"跟得上 leader 的副本集合"。落后太多的副本会被踢出 ISR，此时如果 acks=all 而 ISR 数量不足 min.insync.replicas，生产直接失败，不是降级写入。
| 副本因子 | min.insync.replicas |
可容忍宕机 | 写入可用性 |
|---|---|---|---|
| 3 | 1 | 2 | 高（几乎总可写） |
| 3 | 2 | 1 | 中（推荐设置） |
| 3 | 3 | 0 | 低（任一副本挂即不可写） |
| 2 | 2 | 0 | 低 |
# 查看某个分区的 ISR 现状
kafka-topics.sh --bootstrap-server kafka-1:9092 --describe --topic orders
# Topic 级别设置（也可在 broker 级用 min.insync.replicas 兜底）
kafka-configs.sh --bootstrap-server kafka-1:9092 \
--alter --entity-type topics --entity-name orders \
--add-config min.insync.replicas=2,retention.ms=604800000
NotEnoughReplicasException 与 NotEnoughReplicasAfterAppendException 的区别值得分清：前者是写入前就发现 ISR 不够，消息根本没进日志；后者是已写入 leader 但同步副本数不足，消息可能随后被截断。前者重试安全，后者重试会产生重复。
3. 三个 size 参数：谁能拦住一条消息
max.message.bytes 在 topic 级、message.max.bytes 在 broker 级，两者名字极像，作用点不同：
| 参数 | 作用域 | 拦在哪个环节 | 超限错误 |
|---|---|---|---|
message.max.bytes |
broker | 接收 produce 请求时 | RecordTooLargeException |
max.message.bytes |
topic | 同上，topic 覆盖 broker | RecordTooLargeException |
replica.fetch.max.bytes |
broker | 副本拉取 | 副本永久落后，被踢出 ISR |
fetch.max.bytes |
consumer | 单次 fetch 响应总量 | 不报错，只是分批 |
max.partition.fetch.bytes |
consumer | 单分区单次 fetch | 卡死（见下） |
最隐蔽的一条是 max.partition.fetch.bytes：如果它小于某条消息的大小，消费者每次都拿不到这条消息，又因为"没拿到"而不提交 offset，消费永久卡在这个位置，日志里只有反复的 fetch 请求，没有任何异常。
# broker 端 server.properties（按 Kafka 3.x 核对参数名）
message.max.bytes=1048588
replica.fetch.max.bytes=1048588
# 消费者端必须 >= 生产者可能发出的最大消息
max.partition.fetch.bytes=1048588
fetch.max.bytes=52428800
改这些值要连带改三处：生产者（不设上限）、broker、消费者。只改 broker 会让消费端卡住，只改消费端会让生产端被拒。
4. Rebalance：消费者组的重分配代价
消费组内成员变化、订阅的 topic 分区数变化、心跳超时，都会触发 rebalance（重平衡）。期间整个组停止消费——这是"消费突然停顿几十秒"最常见的原因。
| 协议 | 行为 | 停顿特征 |
|---|---|---|
| Eager（默认，Range/RoundRobin） | 所有成员先放弃全部分区，再重新分配 | 全组停消费 |
| Cooperative（增量协作） | 只迁移需要变动的分区 | 部分分区短暂停 |
# 开启增量协作重平衡（Kafka 2.4+，客户端 2.4+）
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
# 心跳与超时三件套，决定"多久判定成员死亡"
session.timeout.ms=45000
heartbeat.interval.ms=3000
max.poll.interval.ms=300000
max.poll.interval.ms 是最容易踩的：它限制两次 poll() 之间的最大间隔。业务处理慢于这个值，消费者会被判定为死亡，触发 rebalance，然后它处理完回来发现自己已被踢出组——表现为"消息重复消费 + 组内不断重平衡"。
// 手动提交 + 控制单批量的典型骨架
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> r : records) {
handle(r); // 必须快于 max.poll.interval.ms
}
consumer.commitSync(); // 处理完再提交，保证 at-least-once
}
处理慢的正确做法是缩批（max.poll.records）或把重活丢进线程池并自己管 offset，而不是简单调大 max.poll.interval.ms——调大只推迟了判定，没减少处理量。
5. Offset 与重复：至少一次到底重在哪
| 提交时机 | 崩溃后行为 | 结果 |
|---|---|---|
| 处理前提交 | offset 已推进，消息未处理 | 丢消息 |
| 处理后提交 | offset 未推进，消息已处理 | 重复消息 |
| 处理中异步提交 | 两者皆可能 | 不可预测 |
所以"至少一次"不是配置出来的，是提交顺序决定的。要真正去重，幂等键必须落在业务侧：
-- 消费端幂等表，唯一键挡住重复
INSERT INTO consumed_event (event_id, topic, partition_no, offset_no, created_at)
VALUES (?, ?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE event_id = event_id;
只有影响行数为 1（新插入）时才继续业务处理，为 0 说明重复，直接跳过。事务写库与 offset 提交无法原子化，所以业务幂等是唯一可靠解法，事务型生产者只覆盖 Kafka 到 Kafka 的链路。
6. 排错：七种典型故障
| 现象 | 真实原因 | 处理 |
|---|---|---|
RecordTooLargeException |
消息超过 max.message.bytes |
调大 topic 配置，或拆分消息 |
NotEnoughReplicasException |
ISR < min.insync.replicas |
恢复副本 / 临时降 min.insync.replicas |
| 消费卡在固定 offset 不动 | max.partition.fetch.bytes 小于该消息 |
调大该值，或压缩消息体 |
| 组内反复 rebalance | 处理超过 max.poll.interval.ms |
缩 max.poll.records，或改异步处理 |
| 副本持续被踢出 ISR | replica.fetch.max.bytes 太小 / 磁盘慢 |
调大并排查 IO |
| 重复消费量随时间增长 | 处理后才提交，重启回放 | 业务侧幂等表 |
生产端 delivery.timeout.ms 超时 |
重试次数与超时组合不当 | 对齐 request.timeout / retries / 总超时 |
最后一条的关系是固定的：delivery.timeout.ms ≥ linger.ms + request.timeout.ms，且重试总耗时不超它。三者不一致时，生产者会在配置解析阶段就报错。
7. 边界：Kafka 不擅长的事
| 场景 | 更适合的方案 | 原因 |
|---|---|---|
| 单条消息几 MB 的大对象 | 对象存储 + 消息里放引用 | 大消息拖垮页缓存与副本同步 |
| 需要按消息级删除 | 普通队列 / RabbitMQ | Kafka 只能按 offset 或时间过期 |
| 低延迟点对点任务分发 | 专门的任务队列 | 消费组模型面向吞吐，不面向单条确认 |
| 需要复杂路由（按 header 分流） | RabbitMQ exchange | Kafka 只有 topic/partition 两级 |
| 消息顺序 + 全局严格有序 | 单分区，但吞吐受限 | 顺序只在分区内保证 |
分区数也不是越多越好：分区数决定并行度上限，同时增加 rebalance 成本与 broker 的文件句柄压力。常见做法是按目标吞吐倒推分区数，再留一倍余量，而不是一次开到几百。
8. 自测
acks=1到底保证了什么？哪种故障下它会丢消息？- 开启幂等性后，哪三个配置会被强制修改？
max.message.bytes与message.max.bytes的作用域差别是什么？- 消费永久卡在某个 offset 且无异常，先查哪个参数？
- Eager 与 Cooperative 重平衡的关键行为差异是什么？
- 为什么事务型生产者不能替代业务侧幂等？