1.项目地址

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

2.手动提交偏移量

在Springboot整合Kafka中，我们已经了解到了Kafka的基础使用，包括如何发送消息以及接受消息进行消费。
在具体的项目使用中，有时候对于消费者偏移量，需要我们在代码中手动提交，接下来开始讲解如何手动提交偏移量

创建topic

创建一个名为auto-commit-false-topic的Topic，分区数为1
image.png

创建项目

具体创建项目的流程这里不做过多累述，参考Springboot整合Kafka即可

编写配置

重点关注：enable-auto-commit: false

  1. #服务器配置
  2. server:
  3.   port: 8083
  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.       retries: 3
  14.       properties:
  15.         enable.idempotence: true
  16.     consumer:
  17.       group-id: auto-commit-false-group
  18.       enable-auto-commit: false
  19.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  20.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  21.       auto-offset-reset: earliest

编写发送结果处理类

  1. import cn.hutool.log.StaticLog;
  2. import org.apache.kafka.clients.producer.ProducerRecord;
  3. import org.springframework.kafka.core.KafkaProducerException;
  4. import org.springframework.kafka.core.KafkaSendCallback;
  5. import org.springframework.kafka.support.SendResult;
  6. import org.springframework.stereotype.Component;
  8. /**
  9.  * @author: 冯铁城 [17615007230@163.com]
  10.  * @date: 2022-09-07 16:13:57
  11.  * @describe: Kafka发布消息回调处理逻辑实现类
  12.  */
  13. @Component
  14. public class KafkaSendCallBackForString implements KafkaSendCallback<String, String> {
  16.     @Override
  17.     public void onSuccess(SendResult<String, String> result) {
  19.         //1.获取消息属性
  20.         ProducerRecord<String, String> producerRecord = result.getProducerRecord();
  21.         String topic = producerRecord.topic();
  22.         Integer partition = producerRecord.partition();
  23.         String key = producerRecord.key();
  24.         String value = producerRecord.value();
  26.         //2.打印日志
  27.         StaticLog.info(LogTemplateEnum.KAFKA_SEND_SUCCESS_LOG.getTemplate(), topic, partition, key, value);
  28.     }
  30.     @Override
  31.     public void onFailure(KafkaProducerException e) {
  33.         //1.获取消息属性
  34.         ProducerRecord<String, String> failedProducerRecord = e.getFailedProducerRecord();
  35.         String topic = failedProducerRecord.topic();
  36.         Integer partition = failedProducerRecord.partition();
  37.         String key = failedProducerRecord.key();
  38.         String value = failedProducerRecord.value();
  40.         //2.打印日志
  41.         StaticLog.error(LogTemplateEnum.KAFKA_SEND_ERROR_LOG.getTemplate(), topic, partition, key, value);
  43.         //3.异常堆栈信息输出
  44.         e.printStackTrace();
  46.         //4.TODO 可进行自定义的异常逻辑，比如重新发送消息等操作
  47.     }
  48. }
  1. import lombok.AllArgsConstructor;
  2. import lombok.Getter;
  4. /**
  5.  * @author: 冯铁城 [17615007230@163.com]
  6.  * @date: 2022-09-07 16:24:20
  7.  * @describe: 日志模板枚举
  8.  */
  9. @Getter
  10. @AllArgsConstructor
  11. public enum LogTemplateEnum {
  13.     /**
  14.      * Kafka消息发送成功日志模板
  15.      */
  16.     KAFKA_SEND_SUCCESS_LOG("Kafka producer send success! topic:{} partition:{} key:{} value:{}"),
  18.     /**
  19.      * Kafka消息发送失败日志模板
  20.      */
  21.     KAFKA_SEND_ERROR_LOG("Kafka producer send error! topic:{} partition:{} key:{} value:{}"),
  22.     ;
  24.     private final String template;
  25. }

Producer

Producer代码不做过多累述，发送一条消息到对应Topic即可

  1. import com.ftc.autocommitfalse.config.KafkaSendCallBackForString;
  2. import org.junit.jupiter.api.Test;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.boot.test.context.SpringBootTest;
  5. import org.springframework.kafka.core.KafkaTemplate;
  7. @SpringBootTest
  8. class ProducerTests {
  10.     @Autowired
  11.     private KafkaTemplate<String, String> kafkaTemplate;
  13.     @Autowired
  14.     private KafkaSendCallBackForString kafkaSendCallBackForString;
  16.     @Test
  17.     void sendMessage() {
  19.         //1.发送消息
  20.         kafkaTemplate.send("auto-commit-false-topic", "auto-commit-false-message")
  21.                 .addCallback(kafkaSendCallBackForString);
  22.     }
  23. }

Consumer

Consumer代码声明了两个消费者，分别属于不同的消费者组，这样可以保证两个消费者同时消费同一条消息
对于手动提交，包括同步提交和异步提交两种偏移量提交方式

  1. 同步提交：需要指定最大阻塞时间，避免消费者进程长时间阻塞
  2. 异步提交，需要指定callback参数，在提交异常时，根据具体场景进行处理 ```java import cn.hutool.core.util.ObjectUtil; 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;

/**

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

  • @describe: 手动提交偏移量消费者 */ @Component public class Consumer {

    @KafkaListener(topics = {“auto-commit-false-topic”}, groupId = “auto-commit-false-topic-v1”) public void listenerBySyncCommit(ConsumerRecord record, org.apache.kafka.clients.consumer.Consumer consumer) {

    1.  //1.处理消息
    2.  System.out.println("sync receive message:" + record.value());
    4.  //2.同步提交偏移量
    5.  consumer.commitSync(Duration.of(2000, ChronoUnit.MILLIS));

    }

    @KafkaListener(topics = {“auto-commit-false-topic”}, groupId = “auto-commit-false-topic-v2”) public void listenerByAsyncCommit(ConsumerRecord record, org.apache.kafka.clients.consumer.Consumer consumer) {

    1.  //1.处理消息
    2.  System.out.println("async receive message:" + record.value());
    4.  //2.异步提交偏移量
    5.  consumer.commitAsync((offsets, exception) -> {
    6.      if (ObjectUtil.isNotNull(exception)) {
    8.          //3.打印异常信息
    9.          exception.printStackTrace();
    11.          //4.TODO 进行消息重发等机制
    12.      }
    13.  });

    } } ```

    验证

    Producer

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

    Consumer

    如图，消息消费成功image.png
    查看两个消费者的偏移量，偏移量也提交成功
    image.png
    image.png

    3.JSON序列化的消息的发送与接收

    创建Topic

    创建一个名为json-message-topic的Topic
    image.png

    创建项目

    具体创建项目的流程这里不做过多累述，参考Springboot整合Kafka

    编写JSON序列化器/JSON反序列化器

    JSON序列化器

    ```java import cn.hutool.json.JSONObject; import org.apache.kafka.common.serialization.Serializer;

import java.nio.charset.StandardCharsets;

/**

  • @author: 冯铁城 [17615007230@163.com]
  • @date: 2022-09-13 11:27:20

  • @describe: JSON序列化器 */ public class JsonSerialize implements Serializer {

    @Override public byte[] serialize(String topic, JSONObject data) {

    1.  return data.toString().getBytes(StandardCharsets.UTF_8);

    } } ```

    JSON反序列化器

    ```java import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import org.apache.kafka.common.serialization.Deserializer;

import java.nio.charset.StandardCharsets;

/**

  • @author: 冯铁城 [17615007230@163.com]
  • @date: 2022-09-13 11:18:14

  • @describe: JSON反序列化器 */ public class JsonDeserialize implements Deserializer {

    @Override public JSONObject deserialize(String topic, byte[] data) {

    1.  return JSONUtil.parseObj(StrUtil.str(data, StandardCharsets.UTF_8));

    } } ```

    编写配置

    重点关注：value-serializer ```yaml

    服务器配置

    server: port: 8084

spring配置

spring: kafka: bootstrap-servers: 120.48.107.224:9092 producer: acks: -1 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: com.ftc.jsonmessage.config.serialize.JsonSerialize retries: 3 properties: enable.idempotence: true consumer: group-id: json-message-group enable-auto-commit: true key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: com.ftc.jsonmessage.config.serialize.JsonDeserialize auto-offset-reset: earliest

  1. <a name="PRe4L"></a>
  2. ### 编写发送结果处理类
  3. ```java
  4. import cn.hutool.json.JSONObject;
  5. import cn.hutool.log.StaticLog;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.springframework.kafka.core.KafkaProducerException;
  8. import org.springframework.kafka.core.KafkaSendCallback;
  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. public class KafkaSendCallBackForJson implements KafkaSendCallback<String, JSONObject> {
  20.     @Override
  21.     public void onSuccess(SendResult<String, JSONObject> result) {
  23.         //1.获取消息属性
  24.         ProducerRecord<String, JSONObject> producerRecord = result.getProducerRecord();
  25.         String topic = producerRecord.topic();
  26.         Integer partition = producerRecord.partition();
  27.         String key = producerRecord.key();
  28.         JSONObject value = producerRecord.value();
  30.         //2.打印日志
  31.         StaticLog.info(LogTemplateEnum.KAFKA_SEND_SUCCESS_LOG.getTemplate(), topic, partition, key, value.toString());
  32.     }
  34.     @Override
  35.     public void onFailure(KafkaProducerException e) {
  37.         //1.获取消息属性
  38.         ProducerRecord<String, JSONObject> failedProducerRecord = e.getFailedProducerRecord();
  39.         String topic = failedProducerRecord.topic();
  40.         Integer partition = failedProducerRecord.partition();
  41.         String key = failedProducerRecord.key();
  42.         JSONObject value = failedProducerRecord.value();
  44.         //2.打印日志
  45.         StaticLog.error(LogTemplateEnum.KAFKA_SEND_ERROR_LOG.getTemplate(), topic, partition, key, value.toString());
  47.         //3.异常堆栈信息输出
  48.         e.printStackTrace();
  50.         //4.TODO 可进行自定义的异常逻辑，比如重新发送消息等操作
  51.     }
  52. }
  1. import lombok.AllArgsConstructor;
  2. import lombok.Getter;
  4. /**
  5.  * @author: 冯铁城 [17615007230@163.com]
  6.  * @date: 2022-09-07 16:24:20
  7.  * @describe: 日志模板枚举
  8.  */
  9. @Getter
  10. @AllArgsConstructor
  11. public enum LogTemplateEnum {
  13.     /**
  14.      * Kafka消息发送成功日志模板
  15.      */
  16.     KAFKA_SEND_SUCCESS_LOG("Kafka producer send success! topic:{} partition:{} key:{} value:{}"),
  18.     /**
  19.      * Kafka消息发送失败日志模板
  20.      */
  21.     KAFKA_SEND_ERROR_LOG("Kafka producer send error! topic:{} partition:{} key:{} value:{}"),
  22.     ;
  24.     private final String template;
  25. }

Producer

  1. import cn.hutool.json.JSONObject;
  2. import com.ftc.jsonmessage.config.KafkaSendCallBackForJson;
  3. import lombok.SneakyThrows;
  4. import org.junit.jupiter.api.AfterEach;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.boot.test.context.SpringBootTest;
  8. import org.springframework.kafka.core.KafkaTemplate;
  10. import java.util.concurrent.TimeUnit;
  12. @SpringBootTest
  13. class ProducerTests {
  15.     @Autowired
  16.     private KafkaTemplate<String, JSONObject> kafkaTemplate;
  18.     @Autowired
  19.     private KafkaSendCallBackForJson kafkaSendCallBackForJson;
  21.     @AfterEach
  22.     @SneakyThrows({InterruptedException.class})
  23.     void stopSeconds() {
  24.         TimeUnit.MILLISECONDS.sleep(500);
  25.     }
  27.     @Test
  28.     void sendJson() {
  30.         //1.创建消息
  31.         JSONObject message = new JSONObject(true);
  32.         message.set("id", 1);
  33.         message.set("name", "ftc");
  35.         //2.发送消息
  36.         kafkaTemplate.send("json-message-topic", message)
  37.                 .addCallback(kafkaSendCallBackForJson);
  38.     }
  39. }

Consumer

  1. import cn.hutool.json.JSONObject;
  2. import org.apache.kafka.clients.consumer.ConsumerRecord;
  3. import org.springframework.kafka.annotation.KafkaListener;
  4. import org.springframework.stereotype.Component;
  6. /**
  7.  * @author: 冯铁城 [17615007230@163.com]
  8.  * @date: 2022-09-08 16:06:48
  9.  * @describe: JSON数据源监听
  10.  */
  11. @Component
  12. public class JsonConsumer {
  14.     @KafkaListener(topics = {"json-message-topic"}, groupId = "json-message-group")
  15.     public void listener(ConsumerRecord<String, JSONObject> record) {
  16.         System.out.println("JSON数据源_监听获取数据:" + record.value().toString());
  17.     }
  18. }

验证

Producer

消息发送成功
image.png

Consumer

如下，JSON形式消息接收成功
image.png

4.Consumer规则验证

Producer代码

  1. 共发送5条消息，其中第1条消息未指定分区，其余4条消息的分区均为0分区 ```java import com.ftc.single_kafka_producer.config.KafkaSendCallBackForString; import org.apache.kafka.clients.producer.ProducerRecord; 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;

@SpringBootTest class SingleKafkaProducerApplicationTests {

  1. @Autowired
  2. private KafkaTemplate<String, String> kafkaTemplate;
  4. @Autowired
  5. private KafkaSendCallBackForString kafkaSendCallBackForString;
  7. /**
  8.  * 测试主题
  9.  */
  10. private static final String TEST_TOPIC = "test_topic";
  12. /**
  13.  * 测试Key
  14.  */
  15. private static final String TEST_KEY = "test_key";
  17. @Test
  18. void sendMessage() {
  20.     //1.发送消息,指定主题
  21.     kafkaTemplate.send(TEST_TOPIC, "topic_message")
  22.             .addCallback(kafkaSendCallBackForString);
  24.     //2.发送消息,指定主题、Key | 取模之后分区编号=0
  25.     kafkaTemplate.send(TEST_TOPIC, TEST_KEY, "topic_key_message")
  26.             .addCallback(kafkaSendCallBackForString);
  28.     //3.发送消息,指定主题、分区、Key
  29.     kafkaTemplate.send(TEST_TOPIC, 0, TEST_KEY, "topic_partition_key_message")
  30.             .addCallback(kafkaSendCallBackForString);
  32.     //4.发送消息,指定主题、分区、Key、时间戳
  33.     kafkaTemplate.send(TEST_TOPIC, 0, System.currentTimeMillis(), TEST_KEY, "topic_partition_key_timestamp_message")
  34.             .addCallback(kafkaSendCallBackForString);
  36.     //5.发送消息,通过ProducerRecord形式
  37.     ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TEST_TOPIC, 0, TEST_KEY, "record_message");
  38.     kafkaTemplate.send(producerRecord).addCallback(kafkaSendCallBackForString);
  39. }

}

  1. <a name="pfll7"></a>
  2. ### Consumer代码
  4. 1. 共5个消费者
  5. 2. 第1,2个消费者，为同组消费者，同时指定消费同一分区
  6. 3. 第3个消费者，与1,2消费者同组，但消费另一分区
  7. 4. 第4,5个消费者，与1,2,3不同组
  8. ```java
  9. import org.apache.kafka.clients.consumer.ConsumerRecord;
  10. import org.springframework.kafka.annotation.KafkaListener;
  11. import org.springframework.kafka.annotation.TopicPartition;
  12. import org.springframework.stereotype.Component;
  14. /**
  15.  * @author: 冯铁城 [17615007230@163.com]
  16.  * @date: 2022-09-07 18:16:21
  17.  * @describe: 监听者规则测试demo
  18.  */
  19. @Component
  20. public class ConsumerRuleDemo {
  22.     @KafkaListener(topics = {"test_topic"}, groupId = "${spring.kafka.consumer.group-id}",
  23.             topicPartitions = {@TopicPartition(topic = "test_topic", partitions = {"0"})}
  24.     )
  25.     public void listenerDemoV2(ConsumerRecord<String, String> record) {
  26.         System.out.println("receive message from group test_group partition 0 N1:" + record.value());
  27.     }
  29.     @KafkaListener(topics = {"test_topic"}, groupId = "${spring.kafka.consumer.group-id}",
  30.             topicPartitions = {@TopicPartition(topic = "test_topic", partitions = {"0"})}
  31.     )
  32.     public void listenerDemoV3(ConsumerRecord<String, String> record) {
  33.         System.out.println("receive message from group test_group partition 0 N2:" + record.value());
  34.     }
  36.     @KafkaListener(topics = {"test_topic"}, groupId = "${spring.kafka.consumer.group-id}",
  37.             topicPartitions = {@TopicPartition(topic = "test_topic", partitions = {"1"})}
  38.     )
  39.     public void listenerDemoV4(ConsumerRecord<String, String> record) {
  40.         System.out.println("receive message from group test_group partition 1:" + record.value());
  41.     }
  43.     @KafkaListener(topics = {"test_topic"}, groupId = "test_group_bak")
  44.     public void listenerDemoV5(ConsumerRecord<String, String> record) {
  45.         System.out.println("receive message from group test_group_bak N1:" + record.value());
  46.     }
  48.     @KafkaListener(topics = {"test_topic"}, groupId = "test_group_bak")
  49.     public void listenerDemoV6(ConsumerRecord<String, String> record) {
  50.         System.out.println("receive message from group test_group_bak N2:" + record.value());
  51.     }
  52. }

运行结果总结

控制台打印结果如下图：

  1. 第3个消费者消费了第1条消息
  2. 第1,2个消费者，同时消费了第2,3,4,5条消息
  3. 第4个消费者，同时消费了第1条消息
  4. 第5个消费者，同时消费了第2,3,4,5条消息

image.png
由此可以分析出

  1. 同一个消费者组的消费者，在不指定分区时，不会同时消费同一条消息
  2. 同一个消费者组的消费者，在指定分区时，可以同时消费同一条消息
  3. 不同消费者组的消费者，无论是否指定分区，都可以同时消费同一条消息