1.项目地址

https://github.com/GuardFTC/kafka-test.git

2.前言

上一篇文章，我们了解了如何实现Kafka的消息不丢失，但是随之而来的一个问题，就是Kafka消息的重复消费。针对这个问题，我们在本文中进行一个详细的讲解

3.创建Topic

如图，创建一个名为not-repeat-message-topic的Topic
image.png

4.模拟消息重复情况的发生

配置

如下，主要变更为:Consumer调整心跳，session过期，以及最大拉取消息时间间隔
image.png
其中session过期时间要设置在上图的时间范围内

  1. #服务器配置
  2. server:
  3.   port: 8088
  5. #spring配置
  6. spring:
  7.   kafka:
  8.     bootstrap-servers: 120.48.107.224:9092
  9.     producer:
  10.       acks: -1
  11.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  12.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  13.       properties:
  14.         enable.idempotence: true
  15.     consumer:
  16.       group-id: not-repeat-message-group
  17.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  18.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  19.       auto-offset-reset: earliest
  20.       enable-auto-commit: false
  21.       properties:
  22.         heartbeat.interval.ms: 1000
  23.         session.timeout.ms: 7000
  24.         max.poll.interval.ms: 10000

Producer

producer无特殊配置，照常发送消息即可，并且设置异常重试机制

  1. import cn.hutool.core.util.StrUtil;
  2. import cn.hutool.json.JSONUtil;
  3. import cn.hutool.log.StaticLog;
  4. import lombok.RequiredArgsConstructor;
  5. import org.apache.kafka.clients.producer.ProducerRecord;
  6. import org.springframework.kafka.core.KafkaProducerException;
  7. import org.springframework.kafka.core.KafkaSendCallback;
  8. import org.springframework.kafka.core.KafkaTemplate;
  9. import org.springframework.kafka.support.SendResult;
  10. import org.springframework.stereotype.Component;
  12. /**
  13.  * @author: 冯铁城 [17615007230@163.com]
  14.  * @date: 2022-09-07 16:13:57
  15.  * @describe: Kafka发布消息回调处理逻辑实现类
  16.  */
  17. @Component
  18. @RequiredArgsConstructor
  19. public class KafkaSendCallBackForString implements KafkaSendCallback<String, String> {
  21.     private final KafkaTemplate<String, String> kafkaTemplate;
  23.     @Override
  24.     public void onSuccess(SendResult<String, String> result) {
  26.         //1.获取消息属性
  27.         ProducerRecord<String, String> producerRecord = result.getProducerRecord();
  28.         String topic = producerRecord.topic();
  29.         Integer partition = producerRecord.partition();
  30.         String key = producerRecord.key();
  31.         String value = producerRecord.value();
  33.         //2.打印日志
  34.         StaticLog.info(LogTemplateEnum.KAFKA_SEND_SUCCESS_LOG.getTemplate(), topic, partition, key, value);
  35.     }
  37.     @Override
  38.     public void onFailure(KafkaProducerException e) {
  40.         //1.获取消息属性
  41.         ProducerRecord<String, String> failedProducerRecord = e.getFailedProducerRecord();
  42.         String topic = failedProducerRecord.topic();
  43.         Integer partition = failedProducerRecord.partition();
  44.         String key = failedProducerRecord.key();
  45.         String value = failedProducerRecord.value();
  47.         //2.打印日志
  48.         StaticLog.error(LogTemplateEnum.KAFKA_SEND_ERROR_LOG.getTemplate(), topic, partition, key, value);
  50.         //3.异常堆栈信息输出
  51.         e.printStackTrace();
  53.         //4.重发消息
  54.         resendFailMessage(failedProducerRecord);
  55.     }
  57.     /**
  58.      * 消息重发
  59.      *
  60.      * @param producerRecord 消息
  61.      */
  62.     private void resendFailMessage(ProducerRecord<String, String> producerRecord) {
  64.         //1.获取重试次数 TODO 后续可优化为从缓存中获取
  65.         int retryTime = 0;
  67.         //2.获取最大重试次数 TODO 后续可优化为写入配置文件中
  68.         int maxRetryTime = 10;
  70.         //3.重试次数++
  71.         retryTime++;
  73.         //4.判定是否超过限制
  74.         if (retryTime > maxRetryTime) {
  75.             String errorMessage = StrUtil.format(LogTemplateEnum.KAFKA_SEND_OVER_TIME_LIMIT.getTemplate(), JSONUtil.toJsonStr(producerRecord));
  76.             throw new RuntimeException(errorMessage);
  77.         }
  79.         //5.未超过限制重发消息
  80.         kafkaTemplate.send(producerRecord).addCallback(this);
  81.     }
  82. }
  1. import com.ftc.notrepeat.config.KafkaSendCallBackForString;
  2. import lombok.SneakyThrows;
  3. import org.junit.jupiter.api.AfterEach;
  4. import org.junit.jupiter.api.Test;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.boot.test.context.SpringBootTest;
  7. import org.springframework.kafka.core.KafkaTemplate;
  9. import java.util.concurrent.TimeUnit;
  11. @SpringBootTest
  12. class ProducerTests {
  14.     @Autowired
  15.     private KafkaTemplate<String, String> kafkaTemplate;
  17.     @Autowired
  18.     private KafkaSendCallBackForString kafkaSendCallBackForString;
  20.     @AfterEach
  21.     @SneakyThrows(InterruptedException.class)
  22.     void sleep() {
  23.         TimeUnit.SECONDS.sleep(2);
  24.     }
  26.     @Test
  27.     void sendMessage() {
  29.         //1.发送消息
  30.         kafkaTemplate.send("not-repeat-message-topic", "not-repeat-message")
  31.                 .addCallback(kafkaSendCallBackForString);
  32.     }
  33. }

Consumer

  1. Consumer在消费完消息后，线程Sleep30s，Sleep时长一定要大于max.poll.interval.ms和session.timeout.ms
  2. 创建2个消费者，共同属于1个消费者组
  3. 消费完成后手动提交消费者偏移量
  4. 消费者的任务为输出消息 ```java import cn.hutool.core.util.ObjectUtil; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component;

import java.time.Duration; import java.time.temporal.ChronoUnit; import java.util.concurrent.TimeUnit;

/**

  • @author: 冯铁城 [17615007230@163.com]
  • @date: 2022-09-13 10:01:26

  • @describe: 消费者demo */ @Component public class ConsumerDemo {

    @KafkaListener(topics = {“not-repeat-message-topic”}, groupId = “${spring.kafka.consumer.group-id}”) public void consumer1(ConsumerRecord record, Consumer consumer) throws InterruptedException {

    1.  //1.消费消息
    2.  System.out.println("receive message:" + record.value());
    4.  //2.线程睡30s
    5.  TimeUnit.SECONDS.sleep(30);
    7.  //2.手动异步提交偏移量
    8.  consumer.commitAsync((offsets, exception) -> {
    10.      //3.异常不为空,代表提交失败,改为同步提交偏移量,最大阻塞时间2000ms
    11.      if (ObjectUtil.isNotNull(exception)) {
    12.          consumer.commitSync(Duration.of(100, ChronoUnit.MILLIS));
    13.      }
    14.  });

    }

    @KafkaListener(topics = {“not-repeat-message-topic”}, groupId = “${spring.kafka.consumer.group-id}”) public void consumer2(ConsumerRecord record, Consumer consumer) throws InterruptedException {

    1.  //1.消费消息
    2.  System.out.println("receive message:" + record.value());
    4.  //2.线程睡30s
    5.  TimeUnit.SECONDS.sleep(30);
    7.  //2.手动异步提交偏移量
    8.  consumer.commitAsync((offsets, exception) -> {
    10.      //3.异常不为空,代表提交失败,改为同步提交偏移量,最大阻塞时间2000ms
    11.      if (ObjectUtil.isNotNull(exception)) {
    12.          consumer.commitSync(Duration.of(100, ChronoUnit.MILLIS));
    13.      }
    14.  });

    } } ```

    验证

    Producer

    如图，消息发送成功
    image.png
    image.png

    Consumer

    如下图，因为session过期导致触发rebalance机制，致使偏移量提交失败，消息被重复消费
    image.png
    image.png
    通过可视化工具可以看到，消费者偏移量一直提交失败
    image.png

    5.解决方案

    Producer

    Producer对于Kafka消息重复发送的问题，可以通过开启幂等-发送成功数据冗余查询来保证不会重复发送两条消息。上述的验证案例中我们已经开启了Producer幂等，至于数据冗余知道这个概念就好了，本次示例就不写了
    但是我们还是需要对Producer做一些改造，以方便接下来的测试

  1. 发送两批消息，第一批为数字1~4，第二批为数字5
  2. 发送间隔10s，防止这5条消息被Consumer同一批拉取，最终同一批提交偏移量，不便于测试

    代码

    ```java import com.ftc.notrepeat.config.KafkaSendCallBackForString; import lombok.SneakyThrows; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.kafka.core.KafkaTemplate;

import java.util.concurrent.TimeUnit;

@SpringBootTest class ProducerTests {

  1. @Autowired
  2. private KafkaTemplate<String, String> kafkaTemplate;
  4. @Autowired
  5. private KafkaSendCallBackForString kafkaSendCallBackForString;
  7. @AfterEach
  8. @SneakyThrows(InterruptedException.class)
  9. void sleep() {
  10.     TimeUnit.SECONDS.sleep(2);
  11. }
  13. @Test
  14. void sendMessage() throws InterruptedException {
  16.     //1.发送消息
  17.     for (int i = 1; i <= 4; i++) {
  18.         kafkaTemplate.send("not-repeat-message-topic", i + "")
  19.                 .addCallback(kafkaSendCallBackForString);
  20.     }
  22.     TimeUnit.SECONDS.sleep(10);
  23.     kafkaTemplate.send("not-repeat-message-topic", "5")
  24.             .addCallback(kafkaSendCallBackForString);
  25. }

}

  1. <a name="dBE0A"></a>
  2. ### Consumer
  3. <a name="X6W4u"></a>
  4. #### 配置
  5. 对于Consumer的配置，进行如下变更
  7. 1. 适当调长session.timeout.ms时间：7000ms->10000ms
  8. 2. 适当调长max.poll.interval.ms时间：10000ms->15000ms
  9. ```yaml
  10. #spring配置
  11. spring:
  12.   kafka:
  13.     bootstrap-servers: 120.48.107.224:9092
  14.     consumer:
  15.       group-id: not-repeat-message-group
  16.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  17.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  18.       auto-offset-reset: earliest
  19.       enable-auto-commit: false
  20.       properties:
  21.         heartbeat.interval.ms: 1000
  22.         session.timeout.ms: 10000
  23.         max.poll.interval.ms: 15000

代码

对于Consumer的代码，进行如下变更

  1. 对于第一批消息，缩短线程Sleep时间，模拟现实场景中对于业务逻辑代码的执行效率优化
  2. 低于第二批消息，依旧保持线程Sleep时间>session.timeout.ms和max.poll.interval.ms
  3. 通过isConsole来进行控制台输出控制，在输出后，变更isConsole=false，以此来模拟现实场景中在代码层面和数据层面做的幂等处理 ```java import cn.hutool.core.util.ObjectUtil; import lombok.SneakyThrows; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component;

import java.time.Duration; import java.time.temporal.ChronoUnit; import java.util.concurrent.TimeUnit;

/**

  • @author: 冯铁城 [17615007230@163.com]
  • @date: 2022-09-13 10:01:26

  • @describe: 消费者demo */ @Component public class ConsumerDemo {

    @KafkaListener(topics = {“not-repeat-message-topic”}, groupId = “${spring.kafka.consumer.group-id}”) public void consumer1(ConsumerRecord record, Consumer consumer) {

    1.  doMessage(record, consumer);

    }

    @KafkaListener(topics = {“not-repeat-message-topic”}, groupId = “${spring.kafka.consumer.group-id}”) public void consumer2(ConsumerRecord record, Consumer consumer) {

    1.  doMessage(record, consumer);

    }

  1. /**
  2.  * 是否进行控制台打印
  3.  */
  4. boolean isConsole = true;
  6. /**
  7.  * 消费消息
  8.  *
  9.  * @param record   消息
  10.  * @param consumer 消费者，用于手动提交偏移量
  11.  */
  12. @SneakyThrows(InterruptedException.class)
  13. private void doMessage(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
  15.     //1.消费消息
  16.     if (isConsole) {
  17.         System.out.println("receive message:" + record.value());
  18.     } else {
  19.         System.out.println("message is repeat:" + record.value());
  20.     }
  22.     //2.第一批消息睡100ms,第二批消息睡30s
  23.     Integer longTimeMessage = 5;
  24.     if (!longTimeMessage.equals(Integer.parseInt(record.value()))) {
  25.         TimeUnit.MILLISECONDS.sleep(100);
  26.     } else {
  27.         isConsole = false;
  28.         TimeUnit.SECONDS.sleep(30);
  29.     }
  31.     //3.手动异步提交偏移量
  32.     consumer.commitAsync((offsets, exception) -> {
  34.         //4.异常不为空,代表提交失败,改为同步提交偏移量,最大阻塞时间2000ms
  35.         if (ObjectUtil.isNotNull(exception)) {
  36.             consumer.commitSync(Duration.of(2000, ChronoUnit.MILLIS));
  37.         }
  38.     });
  39. }

} ```

验证

Producer

如下图，2批消息全部发送成功
image.png
image.png

Consumer

如下图，第一批消息正常消费，控制台正常输出
image.png
第二批消息，第一次消费正常输出，后续因为线程阻塞，session过期，触发Rebalance机制，导致偏移量提交失败，因此会重复消费消息，但是因为我们代码中进行了幂等处理，所以不会进行常规打印，而是重复消息提示打印
image.png
image.png
由此可以看到，在消费者端进行了相应的配置和处理后，尽最大可能的避免了重复消费的现象，即使当重复消费的现象出现，也可以避免异常数据到达数据层

6.总结

Producer

  1. 开启Producer幂等处理

  2. 可以进行发送数据冗余备份

    Consumer

    规避重复消费

  3. 合理调整heartbeat.interval.ms

  4. 合理调整session.timeout.ms
  5. 合理调整max.poll.interval.ms

  6. 尽可能优化业务逻辑代码执行效率

    当重复消息出现时

  7. 数据层面：通过唯一属性来从数据层面避免重复数据的存在

  8. 代码层面：通过预查询和锁机制，在代码层面确保数据安全性，尽可能保证异常数据不到达数据层