Spring Kafka中如何基于首个消费者返回值动态触发另一个消费者

核心原理 #

Spring 自带的@KafkaListener是静态注解，属性值在应用启动阶段就会解析固定，无法在运行时动态修改注入值。要实现根据第一个消费者返回的参数动态创建第二个消费者，需要使用Spring Kafka提供的KafkaListenerEndpointRegistry和ConcurrentKafkaListenerContainerFactory手动创建、配置、启动监听容器。

实现步骤 #

1. 先注入需要的核心Bean
   在第一个消费者所在的类中注入如下依赖：

@Autowired
private ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerFactory;
@Autowired
private KafkaListenerEndpointRegistry registry;

// 缓存已启动的容器，避免重复创建造成资源浪费
private final ConcurrentHashMap<String, ConcurrentMessageListenerContainer<String, Object>> runningContainers = new ConcurrentHashMap<>();

2. 改造第一个消费者的消费逻辑，收到事件后动态创建第二个消费者

@KafkaListener(topics = "你的调度事件topic", groupId = "scheduler-event-group")
public void consumeSchedulerEvent(SchedulerEvent event, Acknowledgment ack) {
    // 生成容器唯一标识，可自定义规则，比如用topic + filterKey + CustId
    String containerKey = String.format("%s_%s_%s", event.getTopic(), event.getFilterKey(), event.getCustId());
    // 已经启动过的容器直接跳过
    if (runningContainers.containsKey(containerKey)) {
        ack.acknowledge();
        return;
    }

    // 1. 创建监听指定topic的容器
    ConcurrentMessageListenerContainer<String, Object> container = kafkaListenerFactory.createContainer(event.getTopic());
    // 2. 配置容器基础属性
    container.getContainerProperties().setGroupId("dynamic-consumer-group");
    // 如果需要单独的groupId，可以用CustId拼接：container.getContainerProperties().setGroupId("group_" + event.getCustId());

    // 3. 注入动态filterKey，配置过滤策略
    String dynamicFilterKey = event.getFilterKey();
    container.setRecordFilterStrategy(consumerRecord -> {
        // 返回true表示过滤该消息，不进入消费逻辑
        return !dynamicFilterKey.equals(consumerRecord.key());
    });

    // 4. 配置消费逻辑，和你原来的consumeJson逻辑保持一致
    container.setupMessageListener((AcknowledgingMessageListener<String, List<User>>) (data, acknowledgment, headers) -> {
        List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
        List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
        
        // 原有消费逻辑写在这里

        acknowledgment.acknowledge();
    });

    // 5. 启动容器
    container.start();
    runningContainers.put(containerKey, container);
    ack.acknowledge();
}

说明：上面的SchedulerEvent是第一个消费者消费的事件实体类，对应你提到的包含topic、partition、filterKey、CustId字段的JSON结构。

注意事项 #

• 如果你原来的配置中kafkaListenerFactory已经开启了批量消费，上述代码不需要额外修改即可支持批量消费
• 容器生命周期可自主管理：如果需要下线某个动态消费者，直接从runningContainers中取出对应容器调用stop()方法，同时移除缓存即可
• 若不同客户的消费逻辑完全独立，可以给每个动态容器配置独立的groupId，避免偏移量互相影响

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