Spring Boot集成Kafka动态创建消耗者并实现多消耗者发布订阅模子 ...

在Spring Boot集成Kafka时，大家都知道可以使用@KafkaListener注解创建消耗者。但是@KafkaListener注解是静态的，意味着在编译时就已经确定了消耗者，无法动态地创建消耗者。
不过事实上，使用Kafka提供的Java API，使用KafkaConsumer类就可以完成消耗者的动态创建。
我们也知道在一个消耗者组中，同一条消息只会被消耗一次。而动态创建消耗者的情景也通常是满足动态的发布订阅模子（一个发布者，但是大概有不定量的消耗者），以是在这里我们使每个动态创建的消耗者的消耗者组也不一样即可。
下面，我们就来实现一下这个功能。
1，创建消耗者对象

我们可以定义一个“消耗者工厂”类，专门用于创建Kafka消耗者对象，如下：

1. package com.gitee.swsk33.kafkadynamicconsumer.factory;
3. import org.apache.kafka.clients.consumer.ConsumerConfig;
4. import org.apache.kafka.clients.consumer.KafkaConsumer;
5. import org.springframework.beans.factory.annotation.Autowired;
6. import org.springframework.beans.factory.annotation.Value;
7. import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
8. import org.springframework.stereotype.Component;
10. import java.util.Collections;
11. import java.util.Properties;
13. @Component
14. public class KafkaDynamicConsumerFactory {
16.         @Autowired
17.         private KafkaProperties kafkaProperties;
19.         @Value("${spring.kafka.consumer.key-deserializer}")
20.         private String keyDeSerializerClassName;
22.         @Value("${spring.kafka.consumer.value-deserializer}")
23.         private String valueDeSerializerClassName;
25.         /**
26.          * 创建一个Kafka消费者
27.          *
28.          * @param topic   消费者订阅的话题
29.          * @param groupId 消费者组名
30.          * @return 消费者对象
31.          */
32.         public <K, V> KafkaConsumer<K, V> createConsumer(String topic, String groupId) throws ClassNotFoundException {
33.                 Properties consumerProperties = new Properties();
34.                 // 设定一些关于新的消费者的配置信息
35.                 consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
36.                 // 设定新的消费者的组名
37.                 consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
38.                 // 设定反序列化方式
39.                 consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Class.forName(keyDeSerializerClassName));
40.                 consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Class.forName(valueDeSerializerClassName));
41.                 // 新建一个消费者
42.                 KafkaConsumer<K, V> consumer = new KafkaConsumer<>(consumerProperties);
43.                 // 使这个消费者订阅对应话题
44.                 consumer.subscribe(Collections.singleton(topic));
45.                 return consumer;
46.         }
48. }

复制代码

可见这里我们注入了设置文件中反序列化的设置，并用于新创建的消耗者对象。
2，使用定时任务实现消耗者及时订阅

上面仅仅是创建了消耗者，但是消耗者接收消息以及处置处罚消息的操作，也是需要我们手动定义的。
如何让创建的消耗者都去不绝的接收并处置处罚我们的消息呢？大致思路如下：

• 使用定时任务，在定时任务中使消耗者不绝地接收并处置处罚消息
  
• 与此同时，将每个定时任务和消耗者都存起来，后面在消耗者不需要的时候可以移除它们并关闭定时任务
  

这里，我们编写一个上下文类，用于存放所有的消耗者和定时任务，并编写增加和移除定时任务的方法：

1. package com.gitee.swsk33.kafkadynamicconsumer.context;
3. import org.apache.kafka.clients.consumer.ConsumerRecord;
4. import org.apache.kafka.clients.consumer.ConsumerRecords;
5. import org.apache.kafka.clients.consumer.KafkaConsumer;
7. import java.time.Duration;
8. import java.util.Map;
9. import java.util.concurrent.*;
11. /**
12. * Kafka消费者任务上下文
13. */
14. public class KafkaConsumerContext {
16.         /**
17.          * 存放所有自己创建的Kafka消费者任务
18.          * key: groupId
19.          * value: kafka消费者任务
20.          */
21.         private static final Map<String, KafkaConsumer<?, ?>> consumerMap = new ConcurrentHashMap<>();
23.         /**
24.          * 存放所有定时任务的哈希表
25.          * key: groupId
26.          * value: 定时任务对象，用于定时执行kafka消费者的消息消费任务
27.          */
28.         private static final Map<String, ScheduledFuture<?>> scheduleMap = new ConcurrentHashMap<>();
30.         /**
31.          * 任务调度器，用于定时任务
32.          */
33.         private static final ScheduledExecutorService executor = Executors.newScheduledThreadPool(24);
35.         /**
36.          * 添加一个Kafka消费者任务
37.          *
38.          * @param groupId  消费者的组名
39.          * @param consumer 消费者对象
40.          * @param <K>      消息键类型
41.          * @param <V>      消息值类型
42.          */
43.         public static <K, V> void addConsumerTask(String groupId, KafkaConsumer<K, V> consumer) {
44.                 // 创建定时任务
45.                 // 每隔1s拉取消息并处理
46.                 ScheduledFuture<?> future = executor.scheduleAtFixedRate(() -> {
47.                         // 拉取消息
48.                         ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));
49.                         for (ConsumerRecord<K, V> record : records) {
50.                                 // 自定义处理每次拉取的消息
51.                                 System.out.println(record.value());
52.                         }
53.                 }, 0, 1, TimeUnit.SECONDS);
54.                 // 将任务和存入对应的列表以后续管理
55.                 consumerMap.put(groupId, consumer);
56.                 scheduleMap.put(groupId, future);
57.         }
59.         /**
60.          * 移除Kafka消费者定时任务并关闭消费者订阅
61.          *
62.          * @param groupId 消费者的组名
63.          */
64.         public static void removeConsumerTask(String groupId) {
65.                 if (!consumerMap.containsKey(groupId)) {
66.                         return;
67.                 }
68.                 // 取出对应的消费者与任务，并停止
69.                 KafkaConsumer<?, ?> consumer = consumerMap.get(groupId);
70.                 ScheduledFuture<?> future = scheduleMap.get(groupId);
71.                 consumer.close();
72.                 future.cancel(true);
73.                 // 移除列表中的消费者和任务
74.                 consumerMap.remove(groupId);
75.                 scheduleMap.remove(groupId);
76.         }
78. }

复制代码

在增加消耗者定时任务的方法中，调用消耗者对象的poll方法能够拉取一次消息，一次通常大概拉取到多条消息，遍历并处置处罚即可。如许在定时任务中，我们每隔一段时间就拉取一次消息并处置处罚，就实现了消耗者及时订阅消息的效果。
除此之外，在使用定时任务时，即ScheduledExecutorService对象的scheduleAtFixedRate方法，可以实现每隔一定的时间实行一次任务，上述第一个参数传入Runnable接口的实现类，这里使用匿名内部类传入，即自定义的任务，第二个参数是启动延迟时间，第三个参数是每隔多长时间重复实行任务，第四个参数是时间单位。该方法返回一个任务对象，通过这个对象的cancel方法可以取消掉任务。
在移除消耗者定时任务方法中，调用消耗者对象的close方法即可关闭消耗者取消订阅。
3，编写个API测试

现在编写一个API测试一下效果：

1. package com.gitee.swsk33.kafkadynamicconsumer.api;
3. import com.gitee.swsk33.kafkadynamicconsumer.context.KafkaConsumerContext;
4. import com.gitee.swsk33.kafkadynamicconsumer.factory.KafkaDynamicConsumerFactory;
5. import org.apache.kafka.clients.consumer.KafkaConsumer;
6. import org.springframework.beans.factory.annotation.Autowired;
7. import org.springframework.kafka.core.KafkaTemplate;
8. import org.springframework.web.bind.annotation.GetMapping;
9. import org.springframework.web.bind.annotation.PathVariable;
10. import org.springframework.web.bind.annotation.RequestMapping;
11. import org.springframework.web.bind.annotation.RestController;
13. /**
14. * 消息测试api
15. */
16. @RestController
17. @RequestMapping("/api/kafka")
18. public class KafkaTestAPI {
20.         @Autowired
21.         private KafkaTemplate<String, String> kafkaTemplate;
23.         @Autowired
24.         private KafkaDynamicConsumerFactory factory;
26.         @GetMapping("/send")
27.         public String send() {
28.                 kafkaTemplate.send("my-topic", "hello!");
29.                 return "发送完成！";
30.         }
32.         @GetMapping("/create/{groupId}")
33.         public String create(@PathVariable String groupId) throws ClassNotFoundException {
34.                 // 这里统一使用一个topic
35.                 KafkaConsumer<String, String> consumer = factory.createConsumer("my-topic", groupId);
36.                 KafkaConsumerContext.addConsumerTask(groupId, consumer);
37.                 return "创建成功！";
38.         }
40.         @GetMapping("/remove/{groupId}")
41.         public String remove(@PathVariable String groupId) {
42.                 KafkaConsumerContext.removeConsumerTask(groupId);
43.                 return "移除成功！";
44.         }
46. }

复制代码

现在依次访问/api/kafka/create/a和/api/kafka/create/b，就创建了两个消耗者，然后访问/api/kafka/send发送消息，结果如下：

可见，两个消耗者都接收到了消息。
4，总结

可见要动态地创建Kafka消耗者，只需创建并设置好Kafka消耗者对象，并使用定时任务使它们一直拉取消息，就可以实现发布订阅的效果。当然，我们要管理好创建的所有的消耗者和定时任务，防止资源浪费。

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