企业官网建设流程全解析

Kafka 消息零丢失，怎么实现？

一条 Kafka 消息从生产到消费会经过三个阶段：生产、存储、消费。任一阶段都可能丢失。要实现零丢失，就要在三个阶段分别补齐可靠性。

一、先理解 Producer 为什么可能丢消息

Kafka Producer 默认是异步发送。业务线程把消息放入RecordAccumulator就返回，真正发送由 Sender 线程完成。

ProducerRecord<String,String>record=newProducerRecord<>(topic,message);producer.send(record);

这种写法拿不到发送结果，网络抖动、Broker 故障都可能导致消息未写入。

要感知发送结果，应该使用带回调的send：

producer.send(record,newCallback(){@OverridepublicvoidonCompletion(RecordMetadatametadata,Exceptionexception){if(exception!=null){System.err.println("Error sending message: "+exception.getMessage());}else{System.out.println("Message sent successfully: "+metadata.topic()+", "+metadata.partition()+", "+metadata.offset());}}});

二、生产端三板斧

第一板斧：acks=all

acks决定多少副本写入成功后，Producer 才认为消息成功。

• acks=0：不等待确认，吞吐最高，但可能丢消息；
• acks=1：Leader 写入成功即返回；
• acks=all或acks=-1：所有 ISR 副本写入成功才返回，可靠性最高。

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

只设置acks=all还不够，还需要配合min.insync.replicas，避免 ISR 中只剩一个 Leader 时退化成acks=1。

第二板斧：重试

网络抖动、Leader 切换等瞬时故障可以通过重试恢复。

retries=10 retry.backoff.ms=1000

重试次数和重试间隔要结合异常恢复时间设置，避免过早放弃，也不要盲目放大故障。

第三板斧：本地消息表加定时扫描

即使 Broker 返回成功，也可能因为异步刷盘、服务器掉电等原因导致消息最终没有消费。极端严格场景下，需要业务层兜底：

发送前写入本地消息表，状态=待发送 -> 发送 MQ -> 消费成功后更新状态 -> 定时任务扫描过期未确认消息 -> 重新发送

这套方案牺牲吞吐换可靠性，同时必须做好幂等，否则重发会带来重复消费。

三、Broker 端两板斧

第一板斧：严格副本同步

建议配置：

default.replication.factor >= 3 min.insync.replicas >= 2 unclean.leader.election.enable = false

关键约束是：

replication.factor > min.insync.replicas

如果两者相等，只要有一个副本宕机，整个分区就无法满足写入条件。推荐让replication.factor = min.insync.replicas + 1，兼顾可靠性和可用性。

unclean.leader.election.enable=false能防止落后副本成为 Leader 后丢失数据。

第二板斧：严格刷盘

Kafka 和 RocketMQ 一样，默认通过 Page Cache 提升写入性能，再由操作系统异步刷盘。掉电时，尚未刷盘的数据可能丢失。

Kafka 可以调整以下参数：

log.flush.interval.messages log.flush.interval.ms log.flush.scheduler.interval.ms

如果把刷盘间隔设置得非常小，相当于接近同步刷盘。代价是吞吐明显下降，通常会把 Kafka 的吞吐从百万级拉到很低的水平。所以除非业务真的不能接受任何掉电丢失，否则一般优先用多副本保证可靠性。

四、消费端保证零丢失

消费端最容易犯的错误是自动提交 offset。自动提交意味着消息可能还没有真正处理完，offset 已经提交，后续消费者会跳过这些消息。

正确做法是关闭自动提交，业务处理成功后再手动提交。

1. 同步提交

while(true){ConsumerRecords<String,String>records=consumer.poll(Duration.ofSeconds(1));process(records);try{consumer.commitSync();}catch(CommitFailedExceptione){handle(e);}}

commitSync()会阻塞等待提交完成，适合关键场景。

2. 异步提交

consumer.commitAsync((offsets,exception)->{if(exception!=null){handle(exception);}});

异步提交不阻塞，吞吐更高，但需要处理失败回调。

3. 组合提交

常规循环里使用异步提交，退出或异常时用同步提交兜底：

try{while(true){ConsumerRecords<String,String>records=consumer.poll(Duration.ofSeconds(1));process(records);consumer.commitAsync();}}catch(Exceptione){handle(e);}finally{try{consumer.commitSync();}finally{consumer.close();}}

五、零丢失最佳实践

1. 生产端使用producer.send(msg, callback)，不要用单参数send；
2. 生产端设置acks=all；
3. 生产端设置较大的retries；
4. Broker 设置unclean.leader.election.enable=false；
5. Broker 设置replication.factor >= 3；
6. Broker 设置min.insync.replicas > 1；
7. Broker 保证replication.factor > min.insync.replicas；
8. 消费端关闭自动提交，使用commitSync和commitAsync组合提交；
9. 极端业务场景使用本地消息表加定时扫描。

六、总结

Kafka 零丢失不是单点配置，而是贯穿生产、存储、消费的完整方案。生产端要确认加重试，Broker 端要多副本加合理刷盘，消费端要处理成功再提交。

同时要记住：可靠性和吞吐是权衡关系。多副本、acks=all、同步刷盘都会降低吞吐，选型时要根据业务能承受的丢失程度做取舍。