@KafkaListener 为每个监听器单独的过滤逻辑

Question

我需要为监听器工厂生成的每个监听器定义一个自定义过滤策略。 目前，我正在使用 RecordFilterStrategy 来做到这一点：

@Bean
ConcurrentKafkaListenerContainerFactory<String, GenericRecord> kafkaListenerContainerFactoryProject() {
    ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setRecordFilterStrategy(new RecordFilterStrategy<String, GenericRecord>() {
        @Override
        public boolean filter(ConsumerRecord<String, GenericRecord> consumerRecord) {
          return true;
        }
    });
    return factory;
}

但这种过滤适用于该工厂生产的所有侦听器。我需要的是为每个侦听器定义不同的逻辑：

@Component
@SendTo("out")
@KafkaListener(topics = "incoming")
public class TestListener {

    @Filter
    public boolean filter(){
        return true;
    }

    @KafkaHandler
    public TestObject listener(TestObject testObject) {
        log.debug("Received Message: " + testObject);
        return testObject;
    }

}

spring-kafka 是否有一些工具可以做到这一点？还是我需要自己写这样的逻辑？

提前致谢！

Answer 1

不，你没有。您只需要一组带有特定RecordFilterStrategy 的ConcurrentKafkaListenerContainerFactory bean。那么你的@KafkaListener 应该只指定它们基于哪个工厂：

/**
 * The bean name of the {@link org.springframework.kafka.config.KafkaListenerContainerFactory}
 * to use to create the message listener container responsible to serve this endpoint.
 * <p>If not specified, the default container factory is used, if any.
 * @return the container factory bean name.
 */
String containerFactory() default "";