动态创建Kafka消费者组时如何使用RecordFilterStrategy过滤消息

• 方案1：手动包装FilteringMessageListenerAdapter实现过滤
  ConcurrentKafkaListenerContainerFactory内置的setRecordFilterStrategy本质是自动将业务消息监听器包装为FilteringMessageListenerAdapter实例，过滤逻辑完全由该适配器实现，无需依赖工厂类，你只需手动包装自定义监听器即可直接使用RecordFilterStrategy，修改后的代码如下：

Map<String, Object> consumerConfig = ImmutableMap.of(
    BOOTSTRAP_SERVERS_CONFIG, "brokerAddress",
    GROUP_ID_CONFIG, "groupId"
);

DefaultKafkaConsumerFactory<String, String> kafkaConsumerFactory =
        new DefaultKafkaConsumerFactory<>(
                consumerConfig,
                new StringDeserializer(),
                new StringDeserializer());

ContainerProperties containerProperties = new ContainerProperties("topicName");

// 定义过滤策略：返回true表示过滤该消息，不会投递到业务处理逻辑
RecordFilterStrategy<String, String> filterStrategy = record -> {
    // 示例：过滤掉value不包含指定关键字的消息
    return !record.value().contains("需要保留的关键字");
};

// 包装业务监听器
FilteringMessageListenerAdapter<String, String> filteredListener = new FilteringMessageListenerAdapter<>(
    (MessageListener<String, String>) record -> {
        // 你的业务处理逻辑
    },
    filterStrategy
);

containerProperties.setMessageListener(filteredListener);

ConcurrentMessageListenerContainer container =
        new ConcurrentMessageListenerContainer<>(
                kafkaConsumerFactory,
                containerProperties);

container.start();

• 方案2：复用ConcurrentKafkaListenerContainerFactory动态创建容器
  如果你需要用到工厂提供的重试、死信队列等其他扩展能力，也可以直接用工厂动态生成容器，完全支持动态设置消费者组，不需要依赖@KafkaListener注解，示例代码如下：

// 动态构造当前消费者组的配置，可根据业务需求生成不同的groupId
Map<String, Object> consumerConfig = ImmutableMap.of(
    BOOTSTRAP_SERVERS_CONFIG, "brokerAddress",
    GROUP_ID_CONFIG, "动态生成的消费者组ID"
);
DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(
    consumerConfig, new StringDeserializer(), new StringDeserializer()
);

// 构造工厂并设置过滤策略
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.setRecordFilterStrategy(record -> {
    // 过滤逻辑
    return !record.value().contains("需要保留的关键字");
});

// 动态创建容器并启动
ConcurrentMessageListenerContainer<String, String> container = factory.createContainer("topicName");
container.start();

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