spring - @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

不，你没有。您只需要一套ConcurrentKafkaListenerContainerFactory bean 类特别RecordFilterStrategy .那么你的 @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 "";