Spring Boot集成Kafka动态创建消费者与动态删除主题(实现多消费者的发布订阅模型)

简介: Spring Boot集成Kafka动态创建消费者与动态删除主题(实现多消费者的发布订阅模型)

在Spring Boot集成Kafka时,大家都知道可以使用@KafkaListener注解创建消费者。但是@KafkaListener注解是静态的,意味着在编译时就已经确定了消费者,无法动态地创建消费者。

不过事实上,使用Kafka提供的Java API,使用KafkaConsumer类就可以完成消费者的动态创建。

我们也知道在一个消费者组中,同一条消息只会被消费一次。而动态创建消费者的情景也通常是满足动态的发布订阅模型(一个发布者,但是可能有不定量的消费者),所以在这里我们使每个动态创建的消费者的消费者组也不一样即可。

今天我就来分享一下Spring Boot集成Kafka时动态地创建消费者以及动态删除Topic的实现。

1,动态创建消费者

(1) 创建消费者对象

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

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));
        // 设定信任所有类型以反序列化
        consumerProperties.put("spring.json.trusted.packages", "*");
        // 新建一个消费者
        KafkaConsumer<K, V> consumer = new KafkaConsumer<>(consumerProperties);
        // 使这个消费者订阅对应话题
        consumer.subscribe(Collections.singleton(topic));
        return consumer;
    }

}

可见这里我们注入了配置文件中反序列化的配置,并用于新创建的消费者对象。

(2) 使用定时任务实现消费者实时订阅

上面仅仅是创建了消费者,但是消费者接收消息以及处理消息的操作,也是需要我们手动定义的。

如何让创建的消费者都去不停的接收并处理我们的消息呢?大致思路如下:

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

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

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) {
   
        // 先存入消费者以便于后续管理
        consumerMap.put(groupId, consumer);
        // 创建定时任务,每隔1s拉取消息并处理
        ScheduledFuture<?> future = executor.scheduleAtFixedRate(() -> {
   
            // 每次执行拉取消息之前,先检查订阅者是否已被取消(如果订阅者不存在于订阅者列表中说明被取消了)
            // 因为Kafka消费者对象是非线程安全的,因此在这里把取消订阅的逻辑和拉取并处理消息的逻辑写在一起并放入定时器中,判断列表中是否存在消费者对象来确定是否取消任务
            if (!consumerMap.containsKey(groupId)) {
   
                // 取消订阅并关闭消费者
                consumer.unsubscribe();
                consumer.close();
                // 关闭定时任务
                scheduleMap.remove(groupId).cancel(true);
                return;
            }
            // 拉取消息
            ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<K, V> record : records) {
   
                // 自定义处理每次拉取的消息逻辑
                System.out.println(record.value());
            }
        }, 0, 1, TimeUnit.SECONDS);
        // 将任务存入对应的列表以后续管理
        scheduleMap.put(groupId, future);
    }

    /**
     * 移除Kafka消费者定时任务并关闭消费者订阅
     *
     * @param groupId 消费者的组名
     */
    public static void removeConsumerTask(String groupId) {
   
        if (!consumerMap.containsKey(groupId)) {
   
            return;
        }
        // 从列表中移除消费者
        consumerMap.remove(groupId);
    }

}

在增加消费者定时任务的方法中,调用消费者对象的poll方法能够拉取一次消息,一次通常可能拉取到多条消息,遍历并处理即可。这样在定时任务中,我们每隔一段时间就拉取一次消息并处理,就实现了消费者实时订阅消息的效果。

除此之外,在使用定时任务时,即ScheduledExecutorService对象的scheduleAtFixedRate方法,可以实现每隔一定的时间执行一次任务,上述第一个参数传入Runnable接口的实现类,这里使用匿名内部类传入,即自定义的任务,第二个参数是启动延迟时间,第三个参数是每隔多长时间重复执行任务,第四个参数是时间单位。该方法返回一个任务对象,通过这个对象的cancel方法可以取消掉任务。

可见这里,在定时任务中,每次拉取消息之前先判断消费者是否还存在于列表中,以确定消费者是否被取消。为什么要这么操作呢?

因为Kafka的消费者对象是非线程安全的,而ScheduledExecutorService底层使用的是线程池来完成定时任务,如果说我们把取消消费者订阅的逻辑写在另一个方法中,就会导致有两个线程同时操作Kafka消费者,从而抛出异常(定时器线程一直在操作消费者拉取消息,取消订阅又是从定时器之外的线程操作的,这就有两个线程),使得我们不能正常地关闭消费者。(异常内容:kafkaconsumer is not safe for multi-threaded access

所以这里,我把拉取消息逻辑和取消订阅逻辑都写在了一起放在一个定时任务中,使得拉取消息和取消订阅者的操作都是在同一线程(即定时器中线程)执行,而判断是否要取消订阅者的依据就是检查该订阅者是否从订阅者列表中被移除。

(3) 编写个API测试

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

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 "移除成功!";
    }

}

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

image.png

可见,两个消费者都接收到了消息。

2,动态删除Topic

在Spring Boot集成Kafka时,默认情况下向一个Topic发送了消息,若这个Topic不存在则会自动创建。不过如果创建的Topic多了,并且后续不再使用,那会占用服务器资源。

不过,我们也可以通过Kafka库中的AdminClient类实现对Topic的删除。

(1) 配置AdminClient的Bean

创建一个配置类,并在其中配置一个AdminClient类型的Bean,如下:

package com.gitee.swsk33.kafkadynamicconsumer.config;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Configuration
public class KafkaAdminConfig {
   

    /**
     * 读取kafka地址配置
     */
    @Value("${spring.kafka.bootstrap-servers}")
    private String kafkaServerURL;

    /**
     * 注入一个kafka管理实例
     *
     * @return kafka管理对象
     */
    @Bean
    public AdminClient adminClient() {
   
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServerURL);
        return AdminClient.create(properties);
    }

}

可见这里先是从配置文件中读取了配置的Kafka服务器地址,然后通过这个地址配置,创建一个AdminClient对象作为Bean即可。

(2) 使用AdminClient删除

删除也很简单,在需要删除的地方自动装配一个AdminClient对象即可,这里创建一个API试试:

package com.gitee.swsk33.kafkadynamicconsumer.api;

import org.apache.kafka.clients.admin.AdminClient;
import org.springframework.beans.factory.annotation.Autowired;
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;

import java.util.Collections;

@RestController
@RequestMapping("/api/kafka-topic")
public class KafkaTopicAPI {
   

    /**
     * 在需要删除Topic的地方自动装配AdminClient对象
     */
    @Autowired
    private AdminClient adminClient;

    @GetMapping("/delete/{topicId}")
    public String deleteTopic(@PathVariable String topicId) {
   
        adminClient.deleteTopics(Collections.singleton(topicId));
        return "删除Topic完成!";
    }

}

调用AdminClient对象的deleteTopics方法,即可完成删除Topic操作,参数是传入Topic(主题)名称列表,这里只传入一个Topic名,因此使用Collections.singleton方法将这一个名称变成列表形式。

在上述“动态创建消费者”过程中,我们对my-topic主题中发送了消息,因此现在kafka中,有一个my-topic的主题:

image.png

使用kafka自带的kafka-topics.sh脚本就可以查看kafka中的所有Topic,这里kafka是使用Docker部署的,因此上述命令使用docker exec调用的其中的脚本。

好的,现在启动程序后访问接口/api/kafka-topic/delete/my-topic,然后再次查看kafka中的主题:

image.png

可见我们成功地删除了my-topic这个主题。

需要注意的是,如果某个Topic还正在被至少一个消费者订阅着,这个Topic将无法被删除! 所以要删除一个Topic之前请先确保其现在没有被任何消费者订阅。

3,总结

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

上述示例仓库地址:传送门

相关文章
|
11天前
|
Java Spring 容器
Spring系列文章:Spring6集成MyBatis3.5
Spring系列文章:Spring6集成MyBatis3.5
|
11天前
|
安全 Java 测试技术
Spring Boot集成支付宝支付:概念与实战
【4月更文挑战第29天】在电子商务和在线业务应用中,集成有效且安全的支付解决方案是至关重要的。支付宝作为中国领先的支付服务提供商,其支付功能的集成可以显著提升用户体验。本篇博客将详细介绍如何在Spring Boot应用中集成支付宝支付功能,并提供一个实战示例。
48 2
|
2天前
|
Java 关系型数据库 数据库连接
MyBatis-Plus介绍及Spring Boot 3集成指南
MyBatis-Plus是一个MyBatis扩展工具,旨在简化Java开发中的CRUD操作。它具有无侵入性、低损耗、强大的CRUD功能、Lambda表达式支持、主键自动生成、ActiveRecord模式、全局操作和内置代码生成器等特点。在Spring Boot 3中集成MyBatis-Plus,需在pom.xml添加依赖,排除特定版本的mybatis-spring,并用@MapperScan注解指定Mapper接口路径。此外,还介绍了如何使用MyBatis-Plus代码生成器自动生成Mapper、Model、Service和Controller层代码,以加速开发。
35 2
MyBatis-Plus介绍及Spring Boot 3集成指南
|
3天前
|
前端开发 Java 关系型数据库
小唐开始学 Spring Boot——(5)前端显示集成
小唐开始学 Spring Boot——(5)前端显示集成
|
10天前
|
Java 数据库连接 数据安全/隐私保护
springBoot集成token认证,最全Java面试知识点梳理
springBoot集成token认证,最全Java面试知识点梳理
|
11天前
|
NoSQL Java MongoDB
【MongoDB 专栏】MongoDB 与 Spring Boot 的集成实践
【5月更文挑战第11天】本文介绍了如何将非关系型数据库MongoDB与Spring Boot框架集成,以实现高效灵活的数据管理。Spring Boot简化了Spring应用的构建和部署,MongoDB则以其对灵活数据结构的处理能力受到青睐。集成步骤包括:添加MongoDB依赖、配置连接信息、创建数据访问对象(DAO)以及进行数据操作。通过这种方式,开发者可以充分利用两者优势,应对各种数据需求。在实际应用中,结合微服务架构等技术,可以构建高性能、可扩展的系统。掌握MongoDB与Spring Boot集成对于提升开发效率和项目质量至关重要,未来有望在更多领域得到广泛应用。
【MongoDB 专栏】MongoDB 与 Spring Boot 的集成实践
|
11天前
|
消息中间件 JSON Java
RabbitMQ的springboot项目集成使用-01
RabbitMQ的springboot项目集成使用-01
|
11天前
|
搜索推荐 Java 数据库
springboot集成ElasticSearch的具体操作(系统全文检索)
springboot集成ElasticSearch的具体操作(系统全文检索)
|
11天前
|
网络协议 Java Spring
Springboot 集成websocket
Springboot 集成websocket
18 0
|
11天前
|
安全 Java 数据库连接
在IntelliJ IDEA中通过Spring Boot集成达梦数据库:从入门到精通
在IntelliJ IDEA中通过Spring Boot集成达梦数据库:从入门到精通

热门文章

最新文章