关键步骤

设置acksall确保所有ISR副本写入成功。

启用重试retries和幂等性enable.idempotencetrue依赖ProducerId和SequenceNumber。

Broker端

副本数replication.factor≥3ISR最小副本数min.insync.replicas≥2。

使用flush机制定期刷盘通过log.flush.interval.messages配置。

消费者端

手动提交Offsetenable.auto.commitfalse处理完消息后调用commitSync()

Kafka通过生产者端确认机制、Broker端持久化与副本同步、消费者端可靠消费三个核心环节保障消息不丢失。

以下是具体实现机制与步骤

生产者端确保消息成功发送到Broker

acks0生产者不等待Broker确认可能丢失消息不推荐。

acks1Leader副本写入即确认若Leader宕机且未同步到其他副本可能丢失。

acksall或acks-1必须等待所有ISR副本写入成功才返回确认最高可靠性。

重试机制

配置retriesN如3次在Broker临时故障时自动重试。

幂等性enable.idempotencetrue通过Producer

ID和Sequence

每个Partition有多个副本replication.factor≥3Leader处理读写Follower同步数据。

ISRIn-Sync

Replicas只有与Leader保持同步的副本才属于ISR集合。

min.insync.replicas2至少需要2个ISR副本写入成功否则生产者抛出NotEnoughReplicasException。

持久化策略

Cache依赖操作系统缓存加速写入数据异步刷盘。

强制刷盘通过log.flush.interval.messages和log.flush.interval.ms控制刷盘频率高可靠性场景建议启用。

Leader选举与数据恢复

若Leader宕机Controller从ISR中选举新Leader确保数据不丢失。

若所有ISR副本宕机需配置unclean.leader.election.enablefalse禁止非ISR副本成为Leader。

关键源码逻辑

副本同步Leader通过ReplicaFetcherThread向Follower同步数据源码见kafka.server.ReplicaFetcherThread。

ISR管理Broker定期检查Follower的同步状态延迟超过replica.lag.time.max.ms的副本会被移出ISR。

核心机制

关闭自动提交enable.auto.commitfalse在消息处理完成后手动调用commitSync()或commitAsync()。

若消费者崩溃下次启动时从最后提交的Offset恢复避免消息丢失。

事务性消费

结合Kafka事务isolation.levelread_committed仅消费已提交的事务消息。

关键步骤

consumer.poll(Duration.ofMillis(100));for

String

消息发送后等待acksall确认。

若Broker未确认按retries重试。

Broker持久化

Leader和ISR副本将消息写入日志文件。

根据配置决定是否强制刷盘。

消费者消费

处理消息后手动提交Offset。

若消费者崩溃从已提交Offset恢复。

规避措施

正确配置后Kafka可提供至少一次At-Least-Once或精确一次Exactly-Once