可靠性配置：acks、ISR 与事务

Kafka 的"不丢不重"不是一个开关，是生产者、broker、消费者三侧的配置组合。这里把组合表和代价一次讲清。

场景

先明确业务属于哪一档，再选配置。不要为了"更可靠"把所有参数拉满——acks=all + acks 逐条同步会让延迟翻倍。

需求	允许丢	允许重	方案
日志采集	少量可丢	可重	`acks=1`，自动提交
业务事件	不能丢	可重（消费端幂等）	`acks=all` + `min.insync.replicas=2`
支付/订单	不能丢	不能重	`acks=all` + 幂等 + 事务
严格一次（流处理）	不能丢	不能重	事务 + `read_committed`

一、acks 三档

java

props.put(ProducerConfig.ACKS_CONFIG, "all");

值	语义	丢数据场景	延迟
`0`	发出去就不管	网络断、broker 挂，全丢	最低
`1`	leader 写入即返回	leader 写入后、副本同步前挂掉	低
`all`（= `-1`）	ISR 全部确认才返回	几乎不丢（前提 `min.insync.replicas` 配对）	高

acks=all 单独用不保证不丢

acks=all 的含义是"当前 ISR 里所有副本都写成功"。如果 ISR 只剩 leader 一个（其它副本都掉线了），那 leader 一挂，数据照样丢。

必须配合 min.insync.replicas=2：ISR 少于 2 时 broker 直接拒绝写入，报 NOT_ENOUGH_REPLICAS。宁可写失败，也不接受只写了一份还返回成功。

二、min.insync.replicas 怎么配

bash

/opt/kafka/bin/kafka-configs.sh --bootstrap-server 10.0.0.31:9092 \
  --entity-type topics --entity-name order \
  --alter --add-config min.insync.replicas=2

RF	min.insync	容忍故障	说明
3	2	1 台	标准配置
3	3	0 台	任意一台挂就写不进，过于激进
2	2	0 台	挂一台就不可写，只适合双机房
1	1	0 台	单机，没有可靠性可言

组合的行为：

text

RF=3, min.insync=2, acks=all
  ISR=3 → 可写
  ISR=2 → 可写（勉强）
  ISR=1 → 拒绝写入，生产者收到 NOT_ENOUGH_REPLICAS

text

RF=3, min.insync=2, acks=1
  ISR=1 → 仍然可写，且返回成功
  → 这就是"配了等于没配"

三、幂等生产者

网络重试会导致重复。幂等生产者让 broker 按 (producerId, 分区, 序列号) 去重。

java

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);   // 3.0+ 默认 true
props.put(ProducerConfig.ACKS_CONFIG, "all");               // 幂等要求 acks=all
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);

幂等的三个隐含约束（不满足会启动报错）：

acks 必须是 all。
retries 必须 > 0。
max.in.flight.requests.per.connection ≤ 5。

幂等只保证单个生产者会话内、单个分区不重复。生产者重启后 producerId 变，跨会话不去重；跨 topic 也不管。

幂等不解决"应用自己发两次"

幂等是传输层的去重。业务代码里 send() 被调用两次，Kafka 会当成两条正常消息。业务层面的重复要靠业务 key 去重（比如写库时 INSERT ... ON CONFLICT DO NOTHING）。

四、顺序性

Kafka 只保证单分区内有序。要保序：

配置	效果	代价
`max.in.flight.requests.per.connection=1`	严格顺序	吞吐降到 1/5
`enable.idempotence=true` + `max.in.flight=5`	幂等保证顺序	几乎无
按 key 分区	同 key 同分区	热点 key 会倾斜

推荐组合：幂等开 + 5 个 in-flight + 同 key 分区。这是唯一既保序又不牺牲太多吞吐的配置。

重试会打破顺序

max.in.flight > 1 且没开幂等时，第一批失败重试、第二批成功，broker 上的顺序就反了。所以"开重试但不开幂等"是顺序性最常见的坑。

五、事务：真正的 exactly-once

场景：消费 A topic、处理后写 B topic，要求"要么都成功，要么都不生效"。

生产者侧

java

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.0.0.31:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-1");   // 必须稳定且全局唯一

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("order-out", key, value));
    producer.sendOffsetsToTransaction(consumerOffsets, consumer.groupMetadata());
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}

消费者侧

java

props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

read_committed 让消费者看不到未提交的事务消息（读不到超过 LSO 的数据）。read_uncommitted 能看到，但也会看到最终被回滚的消息。

transactional.id 的作用

行为	说明
崩溃恢复	同 ID 的新生产者启动时，会 `initTransactions()` 把旧实例的未完成事务中止
防僵尸	旧实例即使还活着，再写会被 `ProducerFencedException` 拒绝
唯一性	同一个 ID 全局只能有一个活跃生产者，不同实例必须用不同 ID

所以 ID 的规范是 业务名-分区号，比如 order-tx-0、order-tx-1，N 个实例各用一个。

事务有额外开销

事务消息的写入路径比普通消息长（要写事务协调器、两阶段提交）。吞吐会下降 20%~50%，延迟上升。只在真的需要 EOS 时用，别为了"更可靠"无脑开。

六、生产者的超时三件套

java

props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);      // 单次请求等 broker 响应
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 200);          // 重试间隔基数
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);    // 从 send 到最终成功/失败

关系必须满足：

text

delivery.timeout.ms >= request.timeout.ms + linger.ms

否则启动直接报 ConfigException。

实际含义：

request.timeout.ms 只管"一次请求"。
delivery.timeout.ms 管"总共允许多久"，包含所有重试。超时后 send() 的回调收到 TimeoutException，消息可能已写入也可能没有——不确定状态。

七、吞吐与可靠性一起调

java

props.put(ProducerConfig.LINGER_MS_CONFIG, 20);            // 攒 20ms 再发
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);        // 32KB 一批
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd"); // 压缩
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);  // 64MB 缓冲

参数	调大的影响
`linger.ms`	批量更大、吞吐更高、延迟更高
`batch.size`	同上，但对小消息效果有限
`compression.type`	网络和磁盘下降，CPU 上升。`zstd` 综合最优
`buffer.memory`	缓冲更大，但 `send()` 阻塞的阈值也更高

send() 是异步的，缓冲满了才会阻塞。判断是否阻塞：

text

record-size-avg × 未确认条数 接近 buffer.memory  → 开始阻塞

监控 buffer-exhausted-rate 和 buffer-available-bytes。

八、消费者侧不丢

java

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);

java

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        handle(r);                    // 处理成功
    }
    consumer.commitSync();            // 处理完再提交
}

顺序必须是先处理、后提交。反过来（先提交后处理）一旦处理失败，消息就永久丢了。

自动提交 + 异步处理

enable.auto.commit=true 时，提交发生在 poll() 里，它提交的是上一批的位移。如果你的处理是扔进线程池异步做的，会自动提交还没处理完的消息。进程一挂，这些消息全丢。

九、配置检查清单

上线前逐条对：

bash

# broker 侧
/opt/kafka/bin/kafka-configs.sh --bootstrap-server $B --entity-type topics --entity-name order --describe
/opt/kafka/bin/kafka-topics.sh --bootstrap-server $B --describe --topic order

检查项	期望值
`replication.factor`	3（或至少 2）
`min.insync.replicas`	2
`unclean.leader.election.enable`	false
所有分区 `Isr` 长度	= RF
生产者 `acks`	all
生产者幂等	true
消费者提交	手动，处理完再提交
`auto.offset.reset`	`earliest` 或 `latest` 按业务明确

text

Topic: order	PartitionCount: 6	ReplicationFactor: 3	Configs: min.insync.replicas=2
	Topic: order	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2

十、踩过的坑

现象	根因	处置
`NOT_ENOUGH_REPLICAS`	ISR < `min.insync.replicas`	修副本同步，别急着降低 min.insync
生产者 `TimeoutException` 但消息其实写入成功	`delivery.timeout.ms` 到点	用幂等 + 业务去重，不要靠重试兜底
加了重试后顺序错乱	开了重试没开幂等	`enable.idempotence=true`
幂等开着但仍有重复	生产者重启（新 producerId）	业务层去重
事务消费者看不到部分消息	`read_committed` 只能读到 LSO 之前	正常，未提交事务的数据不可见
`ProducerFencedException`	同 `transactional.id` 有另一个实例	每个实例唯一 ID，检查是否重复部署
重启消费者后重复消费大量消息	位移没提交	手动提交，且处理完再提交
改了 topic 配置但客户端行为不变	客户端有自己的 `acks` 等配置	两侧都要对

可靠性配置：acks、ISR 与事务 ​

场景 ​

一、acks 三档 ​

二、min.insync.replicas 怎么配 ​

三、幂等生产者 ​

四、顺序性 ​

五、事务：真正的 exactly-once ​

生产者侧 ​

消费者侧 ​

transactional.id 的作用 ​

六、生产者的超时三件套 ​

七、吞吐与可靠性一起调 ​

八、消费者侧不丢 ​

九、配置检查清单 ​

十、踩过的坑 ​

可靠性配置：acks、ISR 与事务

场景

一、acks 三档

二、min.insync.replicas 怎么配

三、幂等生产者

四、顺序性

五、事务：真正的 exactly-once

生产者侧

消费者侧

transactional.id 的作用

六、生产者的超时三件套

七、吞吐与可靠性一起调

八、消费者侧不丢

九、配置检查清单

十、踩过的坑