2

我需要为监听器工厂生成的每个监听器定义一个自定义过滤策略。目前，我正在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 是否有一些工具可以做到这一点？还是我需要自己编写这样的逻辑？

提前致谢！

4

1 回答 1

2

不，你没有。您只需要一组ConcurrentKafkaListenerContainerFactory具有特定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 "";
于 2020-03-20T14:02:08.243 回答