1.项目地址

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

2.单数据源

创建Topic

通常来说，不建议自动创建主题，因为这样可能会出现很多命名不规范，且完全无意义的主题。因此，接下来的所有例子，我们需要手动创建Topic
如图，创建一个名为single-kafka-topic的Topic，该Topic包含2个Partition。
image.png

添加依赖

  1. <!--Kafka依赖-->
  2. <dependency>
  3.     <groupId>org.springframework.kafka</groupId>
  4.     <artifactId>spring-kafka</artifactId>
  5. </dependency>
  7. <!--web依赖-->
  8. <dependency>
  9.     <groupId>org.springframework.boot</groupId>
  10.     <artifactId>spring-boot-starter-web</artifactId>
  11. </dependency>
  13. <!--lombok依赖-->
  14. <dependency>
  15.     <groupId>org.projectlombok</groupId>
  16.     <artifactId>lombok</artifactId>
  17.     <optional>true</optional>
  18. </dependency>
  20. <!--Hutool依赖-->
  21. <dependency>
  22.     <groupId>cn.hutool</groupId>
  23.     <artifactId>hutool-all</artifactId>
  24.     <version>5.8.6</version>
  25. </dependency>
  27. <!--单元测试依赖-->
  28. <dependency>
  29.     <groupId>org.springframework.boot</groupId>
  30.     <artifactId>spring-boot-starter-test</artifactId>
  31.     <scope>test</scope>
  32. </dependency>

Producer

yml配置

  1. #服务器配置
  2. server:
  3.   port: 8081
  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.         retry.backoff.ms: 100

添加发送结果处理类

  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. }

发布消息

如下，可以看到，发送消息时，可以指定Topic，Partition，Key，Timestamp，或通过ProducerRecord形式发送消息

  1. import com.ftc.single.config.KafkaSendCallBackForString;
  2. import org.apache.kafka.clients.producer.ProducerRecord;
  3. import org.junit.jupiter.api.Test;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.boot.test.context.SpringBootTest;
  6. import org.springframework.kafka.core.KafkaTemplate;
  8. @SpringBootTest
  9. class ProducerTests {
  11.     @Autowired
  12.     private KafkaTemplate<String, String> kafkaTemplate;
  14.     @Autowired
  15.     private KafkaSendCallBackForString kafkaSendCallBackForString;
  17.     /**
  18.      * 测试主题
  19.      */
  20.     private static final String TEST_TOPIC = "single-kafka-topic";
  22.     /**
  23.      * 测试Key
  24.      */
  25.     private static final String TEST_KEY = "single-kafka-key";
  27.     @Test
  28.     void sendMessage() {
  30.         //1.发送消息,指定主题
  31.         kafkaTemplate.send(TEST_TOPIC, "topic_message")
  32.                 .addCallback(kafkaSendCallBackForString);
  34.         //2.发送消息,指定主题、Key | 取模之后分区编号=0
  35.         kafkaTemplate.send(TEST_TOPIC, TEST_KEY, "topic_key_message")
  36.                 .addCallback(kafkaSendCallBackForString);
  38.         //3.发送消息,指定主题、分区、Key
  39.         kafkaTemplate.send(TEST_TOPIC, 0, TEST_KEY, "topic_partition_key_message")
  40.                 .addCallback(kafkaSendCallBackForString);
  42.         //4.发送消息,指定主题、分区、Key、时间戳
  43.         kafkaTemplate.send(TEST_TOPIC, 0, System.currentTimeMillis(), TEST_KEY, "topic_partition_key_timestamp_message")
  44.                 .addCallback(kafkaSendCallBackForString);
  46.         //5.发送消息,通过ProducerRecord形式
  47.         ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TEST_TOPIC, 0, TEST_KEY, "record_message");
  48.         kafkaTemplate.send(producerRecord)
  49.                 .addCallback(kafkaSendCallBackForString);
  50.     }
  51. }

Consumer

yml配置

  1. #服务器配置
  2. server:
  3.   port: 8081
  5. #spring配置
  6. spring:
  7.   kafka:
  8.     bootstrap-servers: 120.48.107.224:9092
  9.     consumer:
  10.       group-id: single-kafka-group
  11.       enable-auto-commit: true
  12.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  13.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  14.       auto-offset-reset: earliest
  15.       auto-commit-interval: 5s

监听消息

如下，在监听消息时，可以指定Topic，GroupId以及Partition

  1. package com.ftc.single.consumer;
  3. import org.apache.kafka.clients.consumer.ConsumerRecord;
  4. import org.springframework.kafka.annotation.KafkaListener;
  5. import org.springframework.kafka.annotation.TopicPartition;
  6. import org.springframework.stereotype.Component;
  8. /**
  9.  * @author: 冯铁城 [17615007230@163.com]
  10.  * @date: 2022-09-13 10:01:26
  11.  * @describe: 消费者demo
  12.  */
  13. @Component
  14. public class Consumer {
  16.     @KafkaListener(topics = {"single-kafka-topic"})
  17.     public void listenerDemo(ConsumerRecord<String, String> record) {
  18.         System.out.println("receive message:" + record.value());
  19.     }
  21.     @KafkaListener(topics = {"single-kafka-topic"}, groupId = "${spring.kafka.consumer.group-id}")
  22.     public void listenerDemoV2(ConsumerRecord<String, String> record) {
  23.         System.out.println("receive message:" + record.value());
  24.     }
  26.     @KafkaListener(topics = {"single-kafka-topic"}, groupId = "${spring.kafka.consumer.group-id}",
  27.             topicPartitions = {@TopicPartition(topic = "single-kafka-topic", partitions = {"0"})}
  28.     )
  29.     public void listenerDemoV3(ConsumerRecord<String, String> record) {
  30.         System.out.println("receive message from partition 0:" + record.value());
  31.     }
  32. }

验证

注意

验证时，需要将消费者代码修改成如下状态，不然会提示偏移量提交异常

  1. package com.ftc.single.consumer;
  3. import org.apache.kafka.clients.consumer.ConsumerRecord;
  4. import org.springframework.kafka.annotation.KafkaListener;
  5. import org.springframework.stereotype.Component;
  7. /**
  8.  * @author: 冯铁城 [17615007230@163.com]
  9.  * @date: 2022-09-13 10:01:26
  10.  * @describe: 消费者demo
  11.  */
  12. @Component
  13. public class Consumer {
  15.     @KafkaListener(topics = {"single-kafka-topic"})
  16.     public void listenerDemo(ConsumerRecord<String, String> record) {
  17.         System.out.println("receive message:" + record.value());
  18.     }
  19. }

消息发送

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

消息接收

如图，消息接收成功
image.png

3.多数据源

创建Topic

如图，创建一个名为multi-kafka-topic的Topic，该Topic包含2个Partition。
image.png

添加依赖

  1. <!--Kafka依赖-->
  2. <dependency>
  3.     <groupId>org.springframework.kafka</groupId>
  4.     <artifactId>spring-kafka</artifactId>
  5. </dependency>
  7. <!--web依赖-->
  8. <dependency>
  9.     <groupId>org.springframework.boot</groupId>
  10.     <artifactId>spring-boot-starter-web</artifactId>
  11. </dependency>
  13. <!--lombok依赖-->
  14. <dependency>
  15.     <groupId>org.projectlombok</groupId>
  16.     <artifactId>lombok</artifactId>
  17.     <optional>true</optional>
  18. </dependency>
  20. <!--Hutool依赖-->
  21. <dependency>
  22.     <groupId>cn.hutool</groupId>
  23.     <artifactId>hutool-all</artifactId>
  24.     <version>5.8.6</version>
  25. </dependency>
  27. <!--单元测试依赖-->
  28. <dependency>
  29.     <groupId>org.springframework.boot</groupId>
  30.     <artifactId>spring-boot-starter-test</artifactId>
  31.     <scope>test</scope>
  32. </dependency>

Producer

yml配置

  1. #服务器配置
  2. server:
  3.   port: 8082
  5. #spring配置
  6. kafka:
  8.   #主数据源配置
  9.   primary:
  10.     producer:
  11.       bootstrap-servers: 120.48.107.224:9092
  12.       ack: "-1"
  13.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  14.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  15.       retries: 3
  16.       idempotence: true
  17.       transaction-id-prefix: primary-transaction-producer-
  19.   #从数据源配置
  20.   secondary:
  21.     producer:
  22.       bootstrap-servers: 120.48.107.224:9092
  23.       ack: "-1"
  24.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  25.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  26.       retries: 3
  27.       idempotence: true

配置类

  1. import lombok.Data;
  3. /**
  4.  * @author: 冯铁城 [17615007230@163.com]
  5.  * @date: 2022-09-08 13:39:47
  6.  * @describe: Producer配置属性
  7.  */
  8. @Data
  9. public class ProducerProperties {
  11.     /**
  12.      * 服务器集群地址
  13.      */
  14.     private String bootstrapServers;
  16.     /**
  17.      * ACK参数
  18.      */
  19.     private String ack;
  21.     /**
  22.      * Key序列化策略类
  23.      */
  24.     private String keySerializer;
  26.     /**
  27.      * Value序列化策略类
  28.      */
  29.     private String valueSerializer;
  31.     /**
  32.      * 重试次数
  33.      */
  34.     private int retries;
  36.     /**
  37.      * 是否开启幂等
  38.      */
  39.     private boolean idempotence;
  41.     /**
  42.      * 事务ID前缀,不为空代表开启Kafka事务
  43.      */
  44.     private String transactionIdPrefix;
  45. }
  1. import org.springframework.boot.context.properties.ConfigurationProperties;
  2. import org.springframework.context.annotation.Bean;
  3. import org.springframework.context.annotation.Configuration;
  4. import org.springframework.context.annotation.Primary;
  6. /**
  7.  * @author: 冯铁城 [17615007230@163.com]
  8.  * @date: 2022-09-08 15:28:45
  9.  * @describe: Producer配置
  10.  */
  11. @Configuration
  12. public class ProducerConfig {
  14.     @Bean(name = "primaryProducerProperties")
  15.     @ConfigurationProperties(prefix = "kafka.primary.producer")
  16.     public ProducerProperties primaryProperties() {
  17.         return new ProducerProperties();
  18.     }
  20.     @Primary
  21.     @Bean(name = "secondaryProducerProperties")
  22.     @ConfigurationProperties(prefix = "kafka.secondary.producer")
  23.     public ProducerProperties secondaryProperties() {
  24.         return new ProducerProperties();
  25.     }
  26. }
  1. import lombok.RequiredArgsConstructor;
  2. import org.apache.kafka.clients.producer.ProducerConfig;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  6. import org.springframework.kafka.core.KafkaTemplate;
  7. import org.springframework.kafka.transaction.KafkaTransactionManager;
  9. import java.util.HashMap;
  10. import java.util.Map;
  12. /**
  13.  * @author: 冯铁城 [17615007230@163.com]
  14.  * @date: 2022-09-08 13:42:00
  15.  * @describe: Producer客户端配置
  16.  */
  17. @Configuration
  18. @RequiredArgsConstructor
  19. public class ProducerClientConfig {
  21.     private final com.ftc.multi.config.producer.ProducerConfig producerConfig;
  23.     @Bean("primaryProducerTemplate")
  24.     public KafkaTemplate<String, String> primaryTemplate() {
  25.         return new KafkaTemplate<>(primaryFactory());
  26.     }
  28.     @Bean("primaryTransactionManager")
  29.     public KafkaTransactionManager<String, String> primaryTransactionManager() {
  30.         return new KafkaTransactionManager<>(primaryFactory());
  31.     }
  33.     @Bean("secondaryProducerTemplate")
  34.     public KafkaTemplate<String, String> secondaryTemplate() {
  35.         return new KafkaTemplate<>(secondaryFactory());
  36.     }
  38.     @Bean("primaryFactory")
  39.     public DefaultKafkaProducerFactory<String, String> primaryFactory() {
  41.         //1.获取配置
  42.         ProducerProperties primaryProperties = producerConfig.primaryProperties();
  44.         //2.获取配置
  45.         Map<String, Object> props = getProps(primaryProperties);
  47.         //3.创建工厂
  48.         DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(props);
  50.         //4.开启事务
  51.         producerFactory.setTransactionIdPrefix(primaryProperties.getTransactionIdPrefix());
  53.         //5.返回
  54.         return producerFactory;
  55.     }
  57.     @Bean("secondaryFactory")
  58.     public DefaultKafkaProducerFactory<String, String> secondaryFactory() {
  60.         //1.获取配置
  61.         ProducerProperties secondaryProperties = producerConfig.secondaryProperties();
  63.         //2.获取配置
  64.         Map<String, Object> props = getProps(secondaryProperties);
  66.         //3.创建工厂,返回
  67.         return new DefaultKafkaProducerFactory<>(props);
  68.     }
  70.     /**
  71.      * 封装不同的配置
  72.      *
  73.      * @param producerProperties Producer属性
  74.      * @return 封装后的配置
  75.      */
  76.     private Map<String, Object> getProps(ProducerProperties producerProperties) {
  78.         //1.生成配置
  79.         Map<String, Object> props = new HashMap<>(10);
  80.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, producerProperties.getBootstrapServers());
  81.         props.put(ProducerConfig.ACKS_CONFIG, producerProperties.getAck());
  82.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, producerProperties.getKeySerializer());
  83.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, producerProperties.getValueSerializer());
  84.         props.put(ProducerConfig.RETRIES_CONFIG, producerProperties.getRetries());
  85.         props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, producerProperties.isIdempotence());
  87.         //2.返回
  88.         return props;
  89.     }
  90. }

添加发送结果处理类

  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. }

发送消息demo

  1. import com.ftc.multi.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.beans.factory.annotation.Qualifier;
  7. import org.springframework.boot.test.context.SpringBootTest;
  8. import org.springframework.kafka.core.KafkaTemplate;
  10. import javax.annotation.Resource;
  11. import java.util.concurrent.TimeUnit;
  13. @SpringBootTest
  14. class ProducerTests {
  16.     @Resource
  17.     @Qualifier("primaryProducerTemplate")
  18.     private KafkaTemplate<String, String> primaryTemplate;
  20.     @Resource
  21.     @Qualifier("secondaryProducerTemplate")
  22.     private KafkaTemplate<String, String> secondaryTemplate;
  24.     @Autowired
  25.     private KafkaSendCallBackForString kafkaSendCallBackForString;
  27.     @AfterEach
  28.     @SneakyThrows({InterruptedException.class})
  29.     void stopSeconds() {
  30.         TimeUnit.MILLISECONDS.sleep(500);
  31.     }
  33.     @Test
  34.     void send() {
  36.         //1.主数据源发送消息
  37.         primaryTemplate.send("multi-kafka-topic",
  38.                 0,
  39.                 "primary",
  40.                 "primary_message"
  41.         ).addCallback(kafkaSendCallBackForString);
  43.         //2.备数据源发送消息
  44.         secondaryTemplate.send(
  45.                 "multi-kafka-topic",
  46.                 1,
  47.                 "secondary",
  48.                 "secondary_message"
  49.         ).addCallback(kafkaSendCallBackForString);
  50.     }
  51. }

Consumer

yml配置

  1. #服务器配置
  2. server:
  3.   port: 8082
  5. #Kafka配置
  6. kafka:
  8.   #主数据源配置
  9.   primary:
  10.     consumer:
  11.       bootstrap-servers: 120.48.107.224:9092
  12.       group-id: multi-kafka-group-primary
  13.       enable-auto-commit: true
  14.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  15.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  16.       auto-offset-reset: earliest
  17.       poll-time: 3000
  19.   #从数据源配置
  20.   secondary:
  21.     consumer:
  22.       bootstrap-servers: 120.48.107.224:9092
  23.       group-id: multi-kafka-group-secondary
  24.       enable-auto-commit: true
  25.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  26.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  27.       auto-offset-reset: earliest
  28.       poll-time: 3000

配置类

  1. import lombok.Data;
  3. /**
  4.  * @author: 冯铁城 [17615007230@163.com]
  5.  * @date: 2022-09-08 19:09:37
  6.  * @describe: Consumer属性
  7.  */
  8. @Data
  9. public class ConsumerProperties {
  11.     /**
  12.      * 服务器集群地址
  13.      */
  14.     private String bootstrapServers;
  16.     /**
  17.      * 消费者组ID
  18.      */
  19.     private String groupId;
  21.     /**
  22.      * 是否自动提交
  23.      */
  24.     private boolean enableAutoCommit;
  26.     /**
  27.      * Key反序列化策略类
  28.      */
  29.     private String keyDeserializer;
  31.     /**
  32.      * Value反序列化策略类
  33.      */
  34.     private String valueDeserializer;
  36.     /**
  37.      * 偏移量消费策略
  38.      */
  39.     private String autoOffsetReset;
  41.     /**
  42.      * 拉取消息超时时间/ms
  43.      */
  44.     private int pollTime;
  45. }
  1. import org.springframework.boot.context.properties.ConfigurationProperties;
  2. import org.springframework.context.annotation.Bean;
  3. import org.springframework.context.annotation.Configuration;
  5. /**
  6.  * @author: 冯铁城 [17615007230@163.com]
  7.  * @date: 2022-09-08 19:14:05
  8.  * @describe: Consumer配置
  9.  */
  10. @Configuration
  11. public class ConsumerConfig {
  13.     @Bean("primaryConsumerProperties")
  14.     @ConfigurationProperties("kafka.primary.consumer")
  15.     public ConsumerProperties primaryProperties() {
  16.         return new ConsumerProperties();
  17.     }
  19.     @Bean("secondaryConsumerProperties")
  20.     @ConfigurationProperties("kafka.secondary.consumer")
  21.     public ConsumerProperties secondaryProperties() {
  22.         return new ConsumerProperties();
  23.     }
  24. }
  1. import lombok.RequiredArgsConstructor;
  2. import org.apache.kafka.clients.consumer.ConsumerConfig;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.kafka.annotation.EnableKafka;
  6. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  7. import org.springframework.kafka.config.KafkaListenerContainerFactory;
  8. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  9. import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
  11. import java.util.HashMap;
  12. import java.util.Map;
  14. /**
  15.  * @author: 冯铁城 [17615007230@163.com]
  16.  * @date: 2022-09-08 16:00:24
  17.  * @describe: Consumer客户端配置
  18.  */
  19. @EnableKafka
  20. @Configuration
  21. @RequiredArgsConstructor
  22. public class ConsumerClientConfig {
  24.     private final com.ftc.multi.config.consumer.ConsumerConfig consumerConfig;
  26.     @Bean("primaryConsumerFactory")
  27.     KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> primaryConsumerFactory() {
  29.         //1.获取属性
  30.         ConsumerProperties consumerProperties = consumerConfig.primaryProperties();
  32.         //2.获取配置
  33.         Map<String, Object> props = getProps(consumerProperties);
  35.         //3.创建工厂
  36.         DefaultKafkaConsumerFactory<Object, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
  38.         //4.创建最终工厂
  39.         ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  40.         factory.setConsumerFactory(consumerFactory);
  41.         factory.getContainerProperties().setPollTimeout(consumerProperties.getPollTime());
  43.         //5.返回
  44.         return factory;
  45.     }
  47.     @Bean("secondaryConsumerFactory")
  48.     KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> secondaryConsumerFactory() {
  50.         //1.获取属性
  51.         ConsumerProperties secondaryProperties = consumerConfig.secondaryProperties();
  53.         //2.获取配置
  54.         Map<String, Object> props = getProps(secondaryProperties);
  56.         //3.创建工厂
  57.         DefaultKafkaConsumerFactory<Object, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
  59.         //4.创建最终工厂
  60.         ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  61.         factory.setConsumerFactory(consumerFactory);
  62.         factory.getContainerProperties().setPollTimeout(secondaryProperties.getPollTime());
  64.         //5.返回
  65.         return factory;
  66.     }
  68.     /**
  69.      * 封装不同的配置
  70.      *
  71.      * @param consumerProperties Consumer属性
  72.      * @return 封装后的配置
  73.      */
  74.     private Map<String, Object> getProps(ConsumerProperties consumerProperties) {
  76.         //.1生成配置
  77.         Map<String, Object> props = new HashMap<>(10);
  78.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, consumerProperties.getBootstrapServers());
  79.         props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerProperties.getGroupId());
  80.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, consumerProperties.isEnableAutoCommit());
  81.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, consumerProperties.getKeyDeserializer());
  82.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, consumerProperties.getValueDeserializer());
  84.         //2.返回
  85.         return props;
  86.     }
  87. }

监听消息demo

  1. import org.apache.kafka.clients.consumer.ConsumerRecord;
  2. import org.springframework.kafka.annotation.KafkaListener;
  3. import org.springframework.kafka.annotation.TopicPartition;
  4. import org.springframework.stereotype.Component;
  6. /**
  7.  * @author: 冯铁城 [17615007230@163.com]
  8.  * @date: 2022-09-08 16:06:48
  9.  * @describe: 多数据源监听
  10.  */
  11. @Component
  12. public class Consumer {
  14.     @KafkaListener(
  15.             topics = {"multi-kafka-topic"},
  16.             containerFactory = "primaryConsumerFactory",
  17.             topicPartitions = {@TopicPartition(topic = "multi-kafka-topic", partitions = {"0"})}
  18.     )
  19.     public void primaryListener(ConsumerRecord<String, String> record) {
  20.         System.out.println("主数据源_监听获取数据:" + record.value());
  21.     }
  23.     @KafkaListener(
  24.             topics = {"multi-kafka-topic"},
  25.             containerFactory = "secondaryConsumerFactory",
  26.             topicPartitions = {@TopicPartition(topic = "multi-kafka-topic", partitions = {"1"})}
  27.     )
  28.     public void secondaryListener(ConsumerRecord<String, String> record) {
  29.         System.out.println("备数据源_监听获取数据:" + record.value());
  30.     }
  31. }

验证

消息发送

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

消息接收

如图，消息接收成功
image.png