如何使用Quarkus订阅Kafka动态生成的主题？

核心思路 #

Quarkus的Kafka客户端支持通过编程方式动态创建消费者，绕过静态配置的限制，直接实现按需订阅运行时生成的主题。

具体实现步骤 #

1. 依赖准备 #

确保项目引入Quarkus Kafka客户端依赖，在pom.xml（Maven）中添加：

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-kafka-client</artifactId>
</dependency>

2. 动态消费者管理类 #

创建一个管理类，负责按需创建、启动和销毁Kafka消费者，避免重复订阅：

import jakarta.enterprise.context.ApplicationScoped;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@ApplicationScoped
public class DynamicKafkaConsumerManager {

    // 存储已创建的消费者，key为主题名
    private final ConcurrentMap<String, Consumer<String, String>> consumerMap = new ConcurrentHashMap<>();

    public void subscribeToTopic(String topic, String bootstrapServers) {
        if (consumerMap.containsKey(topic)) {
            return;
        }

        // 配置消费者核心属性
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "dynamic-group-" + topic);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        // 创建消费者并订阅主题
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(topic));

        // 启动独立消费线程
        new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                consumer.poll(java.time.Duration.ofMillis(100))
                        .forEach(record -> {
                            // 替换为你的业务消息处理逻辑
                            System.out.printf("Topic: %s, Key: %s, Value: %s%n",
                                    record.topic(), record.key(), record.value());
                        });
            }
            consumer.close();
        }).start();

        consumerMap.put(topic, consumer);
    }

    public void unsubscribeFromTopic(String topic) {
        Consumer<String, String> consumer = consumerMap.remove(topic);
        if (consumer != null) {
            consumer.wakeup(); // 中断消费线程并释放资源
        }
    }
}

3. 业务组件调用示例 #

在业务组件中注入管理类，按需触发订阅/取消订阅操作：

import jakarta.inject.Inject;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.QueryParam;

@Path("/kafka/subscription")
public class KafkaSubscriptionController {

    @Inject
    DynamicKafkaConsumerManager consumerManager;

    @POST
    @Path("/subscribe")
    public void subscribe(@QueryParam("topic") String topic, @QueryParam("bootstrap") String bootstrapServers) {
        consumerManager.subscribeToTopic(topic, bootstrapServers);
    }

    @POST
    @Path("/unsubscribe")
    public void unsubscribe(@QueryParam("topic") String topic) {
        consumerManager.unsubscribeFromTopic(topic);
    }
}

关键注意事项 #

• 消费者组隔离：每个动态主题使用独立的消费者组（dynamic-group-{topic}），避免与静态订阅的消费者组冲突。
• 线程安全：用ConcurrentHashMap管理消费者实例，确保多线程环境下的操作安全。
• 资源回收：不再需要订阅时，务必调用取消订阅方法，避免Kafka客户端资源泄漏。
• 配置扩展：可根据业务需求添加SSL、批量消费等自定义配置，将配置参数化提升灵活性。

内容的提问来源于stack exchange，提问作者Van Kuang