Kafka消息可靠性acks、幂等与事务的配置组合与验证Kafka 不会丢消息——这句话成立的前提是你配对了 acks、enable.idempotence 和 transaction.id。三者之间的排列组合有 8 种但只有少数几种能真正保证你期望的可靠性语义。本文逐一验证每种组合的行为差异。一、Kafka 消息可靠性的三层保障Kafka 的消息可靠性由三个独立但相关的机制构成分别作用于生产者、Broker 和消息语义三个层面graph TB subgraph 生产者层 P1[acks 配置br/控制写入确认级别] P2[retries 配置br/控制失败重试次数] P3[enable.idempotencebr/控制幂等投递] end subgraph Broker层 B1[最小 ISR 数量br/min.insync.replicas] B2[副本分布策略br/rack-awareness] B3[日志刷盘策略br/flush.interval] end subgraph 语义层 S1[At-Most-Oncebr/最多一次] S2[At-Least-Oncebr/至少一次] S3[Exactly-Oncebr/精确一次] end P1 -- S1 P1 -- S2 P3 -- S2 P3 -- S3 B1 -- S3 B2 -- S3 style 生产者层 fill:#e3f2fd style Broker层 fill:#fff3e0 style 语义层 fill:#e8f5e9三种交付语义的对比语义保证代价典型场景At-Most-Once不丢但可能丢acks0日志采集、监控指标可容忍少量丢失At-Least-Once不丢但可能重复acksall retries订单创建下游做幂等Exactly-Once不丢也不重复事务 API支付、账务不允许任何偏差二、acks 配置的行为差异与性能影响2.1 acks0发送即成功生产者将消息发送到 socket 缓冲区后立即认为成功不等待 Broker 的任何确认。这是最快的模式但消息可能在网络传输、Broker 写入、Leader 选举等任何环节丢失。/** * acks0 配置示例 - 最大化吞吐、零可靠性 * 适用场景非关键监控数据、日志采集 */ public class FireAndForgetProducer { public static Properties createConfig() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka1:9092,kafka2:9092); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // acks0不等待任何确认 props.put(ProducerConfig.ACKS_CONFIG, 0); // 重试对 acks0 无意义生产者不知道消息是否到达 props.put(ProducerConfig.RETRIES_CONFIG, 0); // 批量发送优化吞吐 props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); props.put(ProducerConfig.LINGER_MS_CONFIG, 5); return props; } }行为特征吞吐量最高约 200K msg/s延迟最低1ms但在网络抖动、Broker 宕机、Leader 选举期间都会丢消息。2.2 acks1Leader 确认即可生产者等待 Leader 副本写入本地日志后返回确认不等待 Follower 同步。行为特征在 Leader 宕机且新 Leader 尚未同步到该消息时可能丢失。假设消息写入 Leader 后立即确认但 Leader 在同步到 Follower 之前宕机新 Leader 不包含此消息。2.3 acksall等待所有 ISR 确认生产者等待所有 ISRIn-Sync Replicas副本都写入后才确认。这是最安全的确认模式。/** * acksall 配置 - 最高可靠性 * 必须配合 min.insync.replicas 使用 */ public class ReliableProducer { public static Properties createConfig() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka1:9092,kafka2:9092,kafka3:9092); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // acksall等待所有 ISR 确认 props.put(ProducerConfig.ACKS_CONFIG, all); // 开启重试配合幂等生产者避免重复 props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100); props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000); // 幂等生产者 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 限制单次请求大小避免在途数据过多 props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); return props; } }2.4 性能对比测试数据在 3 节点 Kafka 集群3 分区、3 副本上的测试结果acks 配置吞吐量 (msg/s)P99延迟 (ms)数据丢失风险acks0215,0001.2Leader 宕机、网络故障acks1180,0003.8Leader 宕机后选举acksall95,00012.5理论不丢需 min.isr≥2acksall 的吞吐量约为 acks0 的 44%这个代价在支付、账务等场景下是必须支付的。三、幂等生产者的序列号机制3.1 幂等性实现原理Kafka 的幂等生产者通过Producer ID (PID) Sequence Number实现。Broker 为每个 PID Partition 组合维护一个 sequence number 计数器收到消息时检查 sequence number 是否连续seq expected→ 正常写入expectedseq expected→ 重复消息跳过返回成功但不写入seq expected→ 消息丢失中间有 gap抛出 OutOfOrderSequenceException/** * 幂等生产者验证 Demo * 演示重复发送被 Broker 自动去重的行为 */ public class IdempotentProducerDemo { private final KafkaProducerString, String producer; private final String topic; public IdempotentProducerDemo(String bootstrapServers, String topic) { this.topic topic; Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); 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); // 幂等生产者要求 max.in.flight.requests ≤ 5 props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); this.producer new KafkaProducer(props); } /** * 发送消息并处理各种异常场景 */ public SendResult sendWithRetry(String key, String value) { ProducerRecordString, String record new ProducerRecord(topic, key, value); int maxRetries 3; int attempt 0; while (attempt maxRetries) { try { RecordMetadata metadata producer.send(record).get(10, TimeUnit.SECONDS); log.info(消息发送成功: topic{}, partition{}, offset{}, key{}, metadata.topic(), metadata.partition(), metadata.offset(), key); return SendResult.success(metadata); } catch (TimeoutException e) { attempt; log.warn(消息发送超时, 重试 {}/{}: key{}, attempt, maxRetries, key); if (attempt maxRetries) { return SendResult.timeout(key, e); } } catch (ExecutionException e) { Throwable cause e.getCause(); // 对于不可重试的异常直接失败 if (cause instanceof RecordTooLargeException) { log.error(消息过大: key{}, size{}, key, record.value().getBytes(StandardCharsets.UTF_8).length); return SendResult.rejected(key, 消息超过大小限制, e); } if (cause instanceof TopicAuthorizationException) { log.error(无权限写入: topic{}, topic); return SendResult.rejected(key, 无写入权限, e); } // OutOfOrderSequenceException序列号异常不可恢复 if (cause instanceof OutOfOrderSequenceException) { log.error(幂等序列号异常, 生产者需要重建: key{}, key, e); // 此异常不可恢复需要关闭当前生产者并重新创建 producer.close(); return SendResult.fatal(key, 幂等性序列异常生产者已关闭, e); } attempt; log.warn(消息发送异常, 重试 {}/{}: key{}, error{}, attempt, maxRetries, key, cause.getMessage()); if (attempt maxRetries) { return SendResult.failed(key, cause); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); return SendResult.interrupted(key); } } return SendResult.failed(key, new RuntimeException(超过最大重试次数)); } public void close() { if (producer ! null) { producer.close(Duration.ofSeconds(10)); } } }3.2 幂等生产者的局限幂等生产者只保证单分区内的去重和单生产者会话内的顺序。如果生产者重启获得新的 PID则无法对新旧 PID 之间的消息做去重。跨 Topic、跨分区的场景需要事务来保证。四、事务的 Exactly-Once 语义4.1 事务的读-处理-写模式Kafka 事务最常见的模式是 consume-transform-produce从源 Topic 消费消息处理后写入目标 Topic整个过程原子化。要么全部成功消费位移和产出消息一起提交要么全部回滚。/** * Kafka 事务生产者 - consume-transform-produce 模式 * 保证从消费到写入的 Exactly-Once 语义 */ Service Slf4j public class TransactionalOrderProcessor { private final KafkaProducerString, String producer; private final KafkaConsumerString, String consumer; private final OrderTransformer transformer; public TransactionalOrderProcessor( Value(${kafka.bootstrap-servers}) String bootstrapServers, OrderTransformer transformer) { this.transformer transformer; // 事务生产者配置 Properties producerProps new Properties(); producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); producerProps.put(ProducerConfig.ACKS_CONFIG, all); producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 事务 ID必须全局唯一通常使用 service-instance 格式 producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, order-processor- getInstanceId()); // 事务超时时间 producerProps.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60000); this.producer new KafkaProducer(producerProps); // 初始化事务 this.producer.initTransactions(); // 消费者配置 Properties consumerProps new Properties(); consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, order-processor-group); consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 事务消费者需要设置隔离级别 consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, read_committed); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); this.consumer new KafkaConsumer(consumerProps); } /** * 事务处理消费 → 转换 → 写入全部原子完成 */ public void processOrderBatch() { consumer.subscribe(List.of(source.orders)); try { while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); if (records.isEmpty()) { continue; } // 开启事务 producer.beginTransaction(); try { // 保存消费位移到事务中 MapTopicPartition, OffsetAndMetadata offsets new HashMap(); for (ConsumerRecordString, String record : records) { try { // 业务处理源消息 → 目标消息 ListProducerRecordString, String outputRecords transformer.transform(record); // 事务内写入目标 Topic for (ProducerRecordString, String output : outputRecords) { producer.send(output, (metadata, exception) - { if (exception ! null) { log.error(事务内写入失败: offset{}, record.offset(), exception); // 不做 abort会在外层统一回滚 throw new RuntimeException( 事务写入失败, exception); } }); } } catch (TransformException e) { // 转换异常记录到死信队列非事务写入 log.error(消息转换失败, 发送到死信: offset{}, key{}, record.offset(), record.key(), e); producer.send(new ProducerRecord( dead.letter.orders, record.key(), TransformError| e.getMessage() )); } // 记录消费位移 TopicPartition tp new TopicPartition( record.topic(), record.partition()); offsets.put(tp, new OffsetAndMetadata(record.offset() 1)); } // 原子提交消费位移 产出消息 producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()); producer.commitTransaction(); log.info(事务提交成功: 处理 {} 条消息, records.count()); } catch (Exception e) { // 业务异常回滚事务 log.error(事务处理异常, 执行回滚: 影响 {} 条消息, records.count(), e); producer.abortTransaction(); // 事务回滚后消息会重新消费At-Least-Once 保证 // 需要确保业务处理逻辑不因为重试产生副作用 } // 事务提交失败可能触发生产者恢复机制 producer.flush(); } } catch (ProducerFencedException e) { // 生产者被隔离可能有同 transaction.id 的另一个实例启动 log.error(生产者被隔离, 可能存在僵尸实例: transactionId{}, getInstanceId(), e); // 执行关闭流程 close(); } catch (Exception e) { log.error(订单处理循环异常, e); close(); } } public void close() { try { producer.close(Duration.ofSeconds(30)); } catch (Exception e) { log.error(生产者关闭异常, e); } try { consumer.close(Duration.ofSeconds(30)); } catch (Exception e) { log.error(消费者关闭异常, e); } } private String getInstanceId() { return System.getenv(HOSTNAME) ! null ? System.getenv(HOSTNAME) : instance- UUID.randomUUID().toString().substring(0, 8); } }4.2 事务的局限性Kafka 事务有三个工程上需要注意的限制跨分区事务是支持的跨 Topic 也是支持的但跨 Kafka 集群不支持事务超时默认 15 分钟transaction.timeout.ms超时后 Broker 自动 abortread_committed隔离级别会过滤未提交的事务消息但读者需要等待事务提交或超时延迟有所增加五、故障注入测试可靠性配置的验证不能止于配置对了就行必须通过故障注入来验证/** * Kafka 消息可靠性故障注入测试 */ SpringBootTest TestInstance(TestInstance.Lifecycle.PER_CLASS) class KafkaReliabilityTest { Test void testAcksAllSurvivesLeaderFailover() throws Exception { // 1. 发送 100 条消息acksall // 2. 在发送过程中 Kill Leader Broker // 3. 等待 Leader 选举完成 // 4. 消费验证100 条消息全部可消费且无丢失 // 预期结果消息数 100无丢失 } Test void testIdempotentProducerNoDuplicatesOnRetry() throws Exception { // 1. 开启幂等生产者 // 2. 对同一 key 发送 10 次相同消息模拟重试 // 3. 消费验证只有 1 条消息写入 // 预期结果仅 1 条消息sequence number 去重生效 } Test void testTransactionalProducerAtomicity() throws Exception { // 1. 在事务中发送 10 条消息到 topic-a // 2. 在第 5 条后模拟异常触发 abortTransaction() // 3. 消费验证topic-a 无新消息全部回滚 // 预期结果零条新增消息事务回滚生效 } }实际压测中的关键验证场景故障类型注入方式预期行为验证指标Leader 宕机kill -9Broker 进程消息不丢失acksall消费数 发送数网络分区iptables 阻断 Broker 间通信生产者重试成功无 TimeoutException 导致丢消息磁盘满fallocate填满日志目录Broker 拒绝写入生产者感知错误非静默丢失消费者 Rebalance启动/停止消费者实例无消息重复消费事务模式下游写入数 源消息数五、总结Kafka 的消息可靠性配置有一个清晰的递进关系acks0只要吞吐、不关心丢失。适用于非关键监控和日志采集。acks1 无幂等比 acks0 好但 Leader 宕机仍可能丢消息。acksall enable.idempotencetrueAt-Least-Once单分区不丢。适用于大多数业务场景配合下游幂等处理即可。acksall 事务 APIExactly-Once跨分区、跨 Topic 的原子性。适用于支付、账务等零容错场景。做可靠性配置时有一个实用原则先在 Broker 侧把min.insync.replicas设为 ≥ 2。这个配置是最后一道防线——如果 ISR 数量不足即使生产者配置了 acksallBroker 也会拒绝写入并抛出NOT_ENOUGH_REPLICAS异常。生产者在收到这个异常后可以根据业务需要决定阻塞等待还是降级处理而不是在副本不足的情况下假装写入成功。