使用Camel-Kafka时，如何避免Kafka主题中的重复未读消息？

先明确你之前的方法未生效的原因：

• 主题压缩（日志压缩）是后台异步清理旧的重复key消息，并非实时拦截重复消息进入队列，所以重复消息会暂时存在队列中，只是后续会被清理，无法满足你「阻止重复未读消息进入」的核心需求。
• 将消息内容设为key结合Kafka单生产者幂等性，只能防止同一个生产者实例重试导致的重复，无法阻止不同Pod的生产者发送的重复消息——每个Pod的生产者是独立实例，幂等性的作用范围仅限单实例。

以下是针对你场景的可行方案：

方案1：分布式实时去重（基于Redis） #

这是最直接满足需求的方案——在生产者发送消息前，通过共享存储校验消息是否已发送过，未发送过才允许进入Kafka队列。

实现思路 #

1. 利用Redis的SETNX（或setIfAbsent）命令做分布式存在性校验：把长数字消息作为唯一key存入Redis，设置一个比消费者最长处理时长更长的过期时间，确保未被处理的消息不会被重复发送。
2. 在Camel路由的生产者端添加前置处理器，完成Redis校验逻辑：校验通过则继续发送到Kafka，不通过则直接丢弃该消息。

Camel Java DSL示例 #

@Autowired
private StringRedisTemplate redisTemplate;

from("your-producer-input-source") // 替换为你的生产者输入源（如定时器、HTTP接口等）
    .process(exchange -> {
        Long messageNum = exchange.getIn().getBody(Long.class);
        String redisKey = "kafka_unique_msg:" + messageNum;
        // 设置过期时间，示例为24小时（根据你的消费者最长处理时间调整）
        Boolean isNewMessage = redisTemplate.opsForValue()
                .setIfAbsent(redisKey, "sent", 24, TimeUnit.HOURS);
        
        if (Boolean.FALSE.equals(isNewMessage)) {
            // 标记为重复消息，停止路由，不发送到Kafka
            exchange.setProperty(Exchange.ROUTE_STOP, Boolean.TRUE);
        }
    })
    .to("kafka:your-target-topic?key={{body}}&bootstrapServers={{kafka.bootstrap.servers}}");

注意事项 #

• 确保Redis采用高可用集群模式，避免单点故障阻塞生产者逻辑。
• 过期时间需合理设置：如果消费者最长处理时间是6小时，可设为8-12小时，既保证未处理消息不会被重复发送，也避免Redis内存占用过高。

方案2：优化Kafka日志压缩（作为补充） #

如果能接受重复消息短暂存在队列，但最终会被清理，可以优化日志压缩配置，加快重复消息的清理速度：

1. 修改主题配置：
   • 设置cleanup.policy=compact（或compact,delete，同时保留过期删除逻辑）
   • 降低min.cleanable.dirty.ratio（比如设为0.3），让Kafka更频繁地执行压缩清理
   • 避免将segment.ms（段文件滚动时间）设置过大，否则压缩触发间隔太长
2. 生产者保持将消息内容作为key发送，Kafka会自动保留每个key的最新消息，旧的重复消息会在后台被清理。

方案3：上游源头去重（如果适用） #

如果你的生产者消息来自有状态的源头（如数据库、上游MQ），可以在源头层面做去重：

• 数据库读取时，用唯一索引约束避免生成重复数据；
• 上游MQ若支持消息去重（如RabbitMQ的幂等队列），先在MQ层面过滤重复消息，再传递给Camel-Kafka生产者。

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