根据标头在反序列化之前过滤消息

Question

有时可以根据标头值在反序列化之前过滤掉消息。使用spring kafka的这个场景是否有任何现有模式。我正在考虑实现类似于 ErrorHandlingDeserializer 除了委托将过滤谓词也作为属性。有什么建议？谢谢。

Answer 1

是的，您可以使用 ErrorHandlingDeserializer 使用的相同技术来返回“标记”对象而不是进行反序列化，然后添加一个 RecordFilterStrategy 过滤具有此类对象的记录，到侦听器（容器工厂当使用@KafkaListener 或为显式侦听器使用过滤适配器时）。

编辑

Spring Boot 和添加过滤器...

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        kafkaConsumerFactory.setRecordFilterStrategy(myFilter());
        return factory;
    }