Spring Boot集成Kafka动态创建消耗者并实现多消耗者发布订阅模子 - IT评测·应用市场-qidao123.com

package com.gitee.swsk33.kafkadynamicconsumer.factory;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.stereotype.Component;
import java.util.Collections;
import java.util.Properties;
@Component
public class KafkaDynamicConsumerFactory {
@Autowired
private KafkaProperties kafkaProperties;
@Value("${spring.kafka.consumer.key-deserializer}")
private String keyDeSerializerClassName;
@Value("${spring.kafka.consumer.value-deserializer}")
private String valueDeSerializerClassName;
/**
* 创建一个Kafka消费者
*
* @param topic 消费者订阅的话题
* @param groupId 消费者组名
* @return 消费者对象
*/
public <K, V> KafkaConsumer<K, V> createConsumer(String topic, String groupId) throws ClassNotFoundException {
Properties consumerProperties = new Properties();
// 设定一些关于新的消费者的配置信息
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
// 设定新的消费者的组名
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
// 设定反序列化方式
consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Class.forName(keyDeSerializerClassName));
consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Class.forName(valueDeSerializerClassName));
// 新建一个消费者
KafkaConsumer<K, V> consumer = new KafkaConsumer<>(consumerProperties);
// 使这个消费者订阅对应话题
consumer.subscribe(Collections.singleton(topic));
return consumer;
}
}

复制代码

package com.gitee.swsk33.kafkadynamicconsumer.context;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.*;
/**
* Kafka消费者任务上下文
*/
public class KafkaConsumerContext {
/**
* 存放所有自己创建的Kafka消费者任务
* key: groupId
* value: kafka消费者任务
*/
private static final Map<String, KafkaConsumer<?, ?>> consumerMap = new ConcurrentHashMap<>();
/**
* 存放所有定时任务的哈希表
* key: groupId
* value: 定时任务对象，用于定时执行kafka消费者的消息消费任务
*/
private static final Map<String, ScheduledFuture<?>> scheduleMap = new ConcurrentHashMap<>();
/**
* 任务调度器，用于定时任务
*/
private static final ScheduledExecutorService executor = Executors.newScheduledThreadPool(24);
/**
* 添加一个Kafka消费者任务
*
* @param groupId 消费者的组名
* @param consumer 消费者对象
* @param <K> 消息键类型
* @param <V> 消息值类型
*/
public static <K, V> void addConsumerTask(String groupId, KafkaConsumer<K, V> consumer) {
// 创建定时任务
// 每隔1s拉取消息并处理
ScheduledFuture<?> future = executor.scheduleAtFixedRate(() -> {
// 拉取消息
ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<K, V> record : records) {
// 自定义处理每次拉取的消息
System.out.println(record.value());
}
}, 0, 1, TimeUnit.SECONDS);
// 将任务和存入对应的列表以后续管理
consumerMap.put(groupId, consumer);
scheduleMap.put(groupId, future);
}
/**
* 移除Kafka消费者定时任务并关闭消费者订阅
*
* @param groupId 消费者的组名
*/
public static void removeConsumerTask(String groupId) {
if (!consumerMap.containsKey(groupId)) {
return;
}
// 取出对应的消费者与任务，并停止
KafkaConsumer<?, ?> consumer = consumerMap.get(groupId);
ScheduledFuture<?> future = scheduleMap.get(groupId);
consumer.close();
future.cancel(true);
// 移除列表中的消费者和任务
consumerMap.remove(groupId);
scheduleMap.remove(groupId);
}
}

复制代码

package com.gitee.swsk33.kafkadynamicconsumer.api;
import com.gitee.swsk33.kafkadynamicconsumer.context.KafkaConsumerContext;
import com.gitee.swsk33.kafkadynamicconsumer.factory.KafkaDynamicConsumerFactory;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 消息测试api
*/
@RestController
@RequestMapping("/api/kafka")
public class KafkaTestAPI {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Autowired
private KafkaDynamicConsumerFactory factory;
@GetMapping("/send")
public String send() {
kafkaTemplate.send("my-topic", "hello!");
return "发送完成！";
}
@GetMapping("/create/{groupId}")
public String create(@PathVariable String groupId) throws ClassNotFoundException {
// 这里统一使用一个topic
KafkaConsumer<String, String> consumer = factory.createConsumer("my-topic", groupId);
KafkaConsumerContext.addConsumerTask(groupId, consumer);
return "创建成功！";
}
@GetMapping("/remove/{groupId}")
public String remove(@PathVariable String groupId) {
KafkaConsumerContext.removeConsumerTask(groupId);
return "移除成功！";
}
}

复制代码