Kafka重复消耗问题剖析及应对策略

Kafka重复消耗问题剖析及应对策略

一、重复消耗问题的产生

Kafka大概发生重复消耗的问题！让我详细解释一下为什么会发生重复消耗，以及如那边置惩罚这个问题。
让我们通过一个具体的场景来理解：
（一）具体时序场景

• 消耗者读取了 offset=100 的消息。
  
• 开始处置惩罚这条消息（好比要耗费 5 秒钟）。
  
• 在处置惩罚过程中（好比第 3 秒时），消耗者进程崩溃了。
  
• 消耗者重启后，由于之前没来得及提交 offset，以是还会从 offset=100 开始消耗。
  

（二）原因分析

这就造成了重复消耗的问题。为什么会如许呢？由于在 Kafka 的设计中，提交 offset 和实际处置惩罚消息是两个独立的操纵，它们之间不是原子性的。
二、解决重复消耗问题的常用策略

（一）实现幂等性处置惩罚

1. public void processOrder(OrderMessage message) {
2.     String orderId = message.getOrderId();
3.     // 使用orderId作为唯一键查询是否处理过
4.     if (orderRepository.existsById(orderId)) {
5.         log.info("订单{}已经处理过，跳过", orderId);
6.         return;
7.     }
8.     // 处理订单
9.     processOrderLogic(message);
10. }

复制代码

（二）使用本地事务

1. @Transactional
2. public void processOrderWithTransaction(OrderMessage message, TopicPartition partition, long offset) {
3.     // 1. 检查消息是否处理过（使用一个专门的表记录已处理的消息）
4.     if (messageProcessRepository.isProcessed(partition, offset)) {
5.         return;
6.     }
7.    
8.     // 2. 处理业务逻辑
9.     processOrderLogic(message);
10.    
11.     // 3. 记录消息已处理
12.     messageProcessRepository.markAsProcessed(partition, offset);
13. }

复制代码

（三）使用分布式事务（更严酷但性能较差）

1. @GlobalTransactional
2. public void processOrderWithGlobalTransaction(OrderMessage message) {
3.     // 1. 处理业务逻辑
4.     processOrderLogic(message);
5.    
6.     // 2. 提交offset
7.     consumer.commitSync();
8. }

复制代码

三、各方案特点及常用方案分析

在实际应用中，最常用的是第一种方案（幂等性处置惩罚），由于它：

• 实现简单，性能好。
  
• 不需要额外的事务支持。
  
• 符合分布式体系的最佳实践。
  

（一）实现幂等性的常见方式

• 使用业务主键：好比订单体系中的订单号。
  
• 使用消息 ID：假如消息自己带有唯一标识。
  
• 使用数据库唯一索引：在插入数据时自动去重。
  
• 使用分布式锁：在处置惩罚之前先获取锁。
  

（二）根据业务场景选择合适方案示例

• 对于订单处置惩罚，使用订单号做幂等。
  
• 对于日记收集，大概允许少量重复。
  
• 对于支付体系，则需要非常严酷的幂等性控制。
  

四、总结

总的来说，重复消耗在分布式体系中是一个普遍的问题，我们不能完全避免它，但可以通过合适的设计来确保它不会影响业务的正确性。这就是为什么“幂等性”在分布式体系设计中云云紧张。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。