Kafka数据重复和数据丢失的解决方案

痛定思痛。 2023-10-04 17:46 64阅读 0赞

**1.Kafka数据重复的解决方案：**

对每个生产者生成的每条数据，都添加由生产者id，分区号，随机的序列号组成的标识符： (producerid,partition,SequenceId),通过标识符对数据进行去重。

**2.Kafka数据丢失的解决方案：**

1.  设置enable.auto.commit=false，每次处理完手动提交，确保消息真的被消费并处理完成。[Kafka的消息自动提交和手动提交\_刘Java的博客-CSDN博客\_kafka 自动提交][Kafka_Java_-CSDN_kafka]
2.  配置上消息重试的机制。[Spring-Kafka —— 消费重试机制实现 - 曹伟雄 - 博客园][Spring-Kafka _ _ - _ -]

**Kafka自动提交和手动提交说明：**

使用原始**apache-kafka**依赖的**API**来消费数据：

1.  如果**enable.auto.commit**为**true**，则表示自动提交，但不会在拉取数据之后立即提交。在一次**poll**的数据处理完毕之后，将会在下一次**poll**数据的时候，首先检查是否到达了**auto.commit.interval.ms**自动提交间隔的时间，如果到达了（默认5s），那么会提交此前拉取的消息的最大偏移量，否则不会提交。
2.  如果**enable.auto.commit**为**false**，则表示手动提交，那么需要通过**consumer.commitAsync()**或者**commitSync()**手动提交偏移量，这两个方法将会提交目前最大的**offset**，否则重启之后将会消费此前的数据。

使用**spring-kafka**的**@Listener**注解来消费数据：

1.  如果**enable.auto.commit**为**true**，则表示自动提交，但不会在拉取数据之后立即提交。在一次**poll**的数据处理完毕之后，将会在下一次**poll**数据的时候，首先检查是否到达了**auto.commit.interval.ms**自动提交间隔的时间，如果到达了（默认5s），那么会提交此前拉去的消息的最大偏移量，否则不会提交。
2.  如果**enable.auto.commit**为**false**，则表示手动提交，此时需要注意选择提交的模式AckMode。

AckMode模式可以在工厂类配置：

@Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualListenerContainerFactory(
    	ConsumerFactory<String, String> consumerFactory) {
    
    	ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    	factory.setConsumerFactory(consumerFactory);
    	factory.getContainerProperties().setPollTimeout(1500);
    	factory.setBatchListener(true);
    	//配置手动提交offset，默认BATCH
    	factory.getContainerProperties().setAckMode(AckMode.BATCH);
    	return factory;
    }

<table style="width:500px;"> 
 <thead> 
  <tr> 
   <th style="width:111px;">AckMode模式</th> 
   <th style="width:388px;">作用</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="width:111px;">BATCH</td> 
   <td style="width:388px;"><strong>默认的提交模式</strong>。当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交，由Spring帮我们提交。</td> 
  </tr> 
  <tr> 
   <td style="width:111px;">RECORD</td> 
   <td style="width:388px;">当每一条记录被消费者监听器（ListenerConsumer）处理之后提交，由Spring帮我们提交。</td> 
  </tr> 
  <tr> 
   <td style="width:111px;">TIME</td> 
   <td style="width:388px;">当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交，由Spring帮我们提交。</td> 
  </tr> 
  <tr> 
   <td style="width:111px;">COUNT</td> 
   <td style="width:388px;">当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交，由Spring帮我们提交。</td> 
  </tr> 
  <tr> 
   <td style="width:111px;">COUNT_TIME</td> 
   <td style="width:388px;">TIME和COUNT有一个条件满足时提交，由Spring帮我们提交。</td> 
  </tr> 
  <tr> 
   <td style="width:111px;">MANUAL</td> 
   <td style="width:388px;">需要对监听消息的方法中引入 Acknowledgment参数，并在代码中调用acknowledge()方法进行手动提交。实际上，对于每一批poll()的数据，每次调用acknowledge()方法之后仅仅是将offset存放到本地map缓存，在下一次poll的时候，在poll新数据之前从缓存中拿出来批量提交，也就是说与BATCH有相同的语义。</td> 
  </tr> 
  <tr> 
   <td style="width:111px;">MANUAL_IMMEDIATE</td> 
   <td style="width:388px;">需要对监听消息的方法中引入 Acknowledgment参数，并在代码中调用acknowledge()方法进行手动提交。实际上，对于每一批poll()的数据，每次调用acknowledge()方法之后立即进行偏移量的提交。</td> 
  </tr> 
 </tbody> 
</table>

*  由于默认的提交模式是**BATCH**，因此在使用**@Listener**注解来消费数据时，即使**enable.auto.commit**为**false**，偏移量也会在每一批**poll()**的数据被消费者监听器**（ListenerConsumer）**处理之后提交，这里的提交实际上是手动提交，但是这个“手动提交”操作由**Spring**帮我们做了，因此如果不设置**AckMode**为**MANUAL**或者**MANUAL\_IMMEDIATE**，我们仍然会觉得这些数据被“自动提交”了，实际上是由**Spring**帮我们执行了手动提交的代码，造成误解。
 *  **MANUAL**和**MANUAL\_IMMEDIATE**的区别是：**MANUAL\_IMMEDIATE**是消费完一个消息就提交，**MANUAL**是处理完一批消息（默认500）之后，在下一次拉取消息之前批量提交。
 *  如果中间有一批数据没有提交，那么在一次消费过程中，这些没有提交的数据不会重复消费，而是会一直向后消费，除非重启消费者，会被再次消费。如果后面有消息的offset被提交，那么该offset之前的所有消息都算作已提交，重启之后也不会被再次消费。

[Kafka_Java_-CSDN_kafka]: https://blog.csdn.net/weixin_43767015/article/details/122005747?spm=1001.2101.3001.6661.1&utm_medium=distribute.pc_relevant_t0.none-task-blog-2~default~CTRLIST~Rate-1.pc_relevant_default&depth_1-utm_source=distribute.pc_relevant_t0.none-task-blog-2~default~CTRLIST~Rate-1.pc_relevant_default&utm_relevant_index=1
[Spring-Kafka _ _ - _ -]: https://www.cnblogs.com/caoweixiong/p/11181587.html