Quarkus 2中无需@Incoming注解动态创建独立Kafka消费者

核心思路 #

利用SmallRye Reactive Messaging的通道模板统一配置，结合Quarkus启动事件监听动态绑定通用处理函数，同时为每个通道配置独立线程池保证隔离性。

步骤1：配置通道模板 #

将所有Kafka消费通道的通用配置抽成模板，新增通道时只需继承模板并指定专属主题，无需重复编写通用配置：

mp:
  messaging:
    incoming:
      # 定义通用模板（标记为template: true）
      kafka-consumer-template:
        connector: smallrye-kafka
        bootstrap.servers: ${kafka.bootstrap.servers}
        group.id: kafka-dynamic-group
        auto.offset.reset: earliest
        template: true
        # 为每个通道配置独立线程池，使用${channel}变量自动替换通道名
        thread-pool:
          name: kafka-pool-${channel}
          size: 8
          max-size: 16

      # 现有通道：继承模板
      default:
        template: kafka-consumer-template
        topics: topicDefault1, topicDefault2

      # 新增通道1：仅需指定主题
      specificA:
        template: kafka-consumer-template
        topics: topicSpecificA1, topicSpecificA2

      # 新增通道2：仅需指定主题
      specificB:
        template: kafka-consumer-template
        topics: topicSpecificB1, topicSpecificB2

步骤2：动态绑定通用处理函数 #

通过Quarkus启动事件监听，扫描所有Kafka消费通道并绑定通用处理逻辑，无需为每个通道编写单独的@Incoming方法：

import io.smallrye.reactive.messaging.ChannelRegistry;
import io.smallrye.reactive.messaging.Message;
import io.smallrye.reactive.messaging.kafka.IncomingKafkaRecordMetadata;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.event.Observes;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;
import java.util.concurrent.CompletionStage;

@ApplicationScoped
public class DynamicKafkaMessageHandler {

    private static final Logger LOG = Logger.getLogger(DynamicKafkaMessageHandler.class);

    @Inject
    ChannelRegistry channelRegistry;

    // 应用启动时自动注册所有Kafka通道的处理函数
    void onApplicationStart(@Observes StartupEvent event) {
        channelRegistry.getChannels().stream()
                // 过滤出使用smallrye-kafka连接器的入站通道
                .filter(channel -> {
                    String connector = channelRegistry.getChannelConfiguration(channel.getName())
                            .getOptionalValue("connector", String.class)
                            .orElse("");
                    return "smallrye-kafka".equals(connector);
                })
                .forEach(channel -> {
                    String channelName = channel.getName();
                    LOG.infof("绑定通用处理函数到通道: %s", channelName);
                    // 订阅通道消息到通用处理逻辑
                    channel.getSubscriber().subscribe().with(this::processMessage);
                });
    }

    // 通用消息处理函数
    private CompletionStage<Void> processMessage(Message<OhpEntry> message) {
        try {
            // 获取消息所属主题（可选，用于区分业务逻辑）
            String topic = message.getMetadata(IncomingKafkaRecordMetadata.class)
                    .map(IncomingKafkaRecordMetadata::getTopic)
                    .orElse("unknown-topic");

            // 执行你的业务逻辑
            handleBusinessLogic(message.getPayload(), topic);

            // 消息确认
            return message.ack();
        } catch (Exception e) {
            LOG.errorf("处理消息失败（主题：%s）", 
                    message.getMetadata(IncomingKafkaRecordMetadata.class)
                            .map(IncomingKafkaRecordMetadata::getTopic)
                            .orElse("unknown-topic"), e);
            // 消息拒绝（触发重试或死信队列）
            return message.nack(e);
        }
    }

    // 自定义业务逻辑
    private void handleBusinessLogic(OhpEntry payload, String topic) {
        // 这里写你的业务处理代码
        LOG.infof("处理来自主题 %s 的消息: %s", topic, payload);
    }
}

步骤3：确保通道处理完全独立 #

通过配置中的thread-pool.name: kafka-pool-${channel}，每个通道会使用专属线程池，避免因某一通道处理耗时过长或发送nack而影响其他通道。

如果需要更精细的线程控制，也可以在处理函数中手动指定线程池，例如结合Quarkus的Executor注入实现动态线程池绑定。

新增通道的方式 #

后续新增specificX类主题时，只需在配置文件中添加一段继承模板的配置即可：

mp:
  messaging:
    incoming:
      specificX:
        template: kafka-consumer-template
        topics: topicSpecificX1, topicSpecificX2

无需修改任何代码，应用重启后会自动绑定通用处理函数并使用独立线程池。

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