Spring Cloud Stream Kafka函数式消费者重复消息检测优化方案问询

当前实现代码 #

@Component
public class MessageSubscriber {

    @Autowired
    MessageConsumer messageConsumer;

    public Consumer<Message<PlaneEvent>> planeEventConsumer() {
        return event -> messageConsumer.consumePlaneEvent(event);
    }

    @Component
    public static class MessageConsumer {
        
        @Transactional
        @DuplicateCheck(key = "id")
        public void consumePlaneEvent(Message<PlaneEvent> msg) {
            // 业务逻辑处理
        }

    }
}

现有重复检测机制 #

自定义@DuplicateCheck注解通过Spring AOP拦截消费方法，利用注解指定的key，结合Message中的kafka_receivedTopic头信息，在数据库中完成重复消息的检测。

需要优化的问题点 #

• 拦截函数式消费者：了解ChannelInterceptor但不愿采用，因为它会拦截所有通道，需要强制规范通道命名，还会提升跳过重复消息的处理复杂度，希望用非侵入式方案解决。
• 获取Kafka主题或绑定目标：依赖kafka_receivedTopic头信息的方式不够稳定，灵活性欠缺。
• 函数式消费者事务处理：由于@Transactional注解会生成动态代理，必须把消费逻辑封装到独立类并创建Bean，操作远不如旧版@StreamListener注解式消费者简便。

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