RabbitMQ避免重复消费

在Java中，可以使用消息队列来实现消息的异步处理，其中常用的消息队列有 RabbitMQ、ActiveMQ、Kafka 等。

什么是幂等性？

幂等性是指无论操作执行多少次，都是得到相同的结果，而不会产生其他副作用。

在rabbitMQ中

什么是消息重复消费？

同一条消息在MQ中被消费多次

出现重复消费的原因：

生产者发送一条消息到rabbitMQ，但rabbitMQ尚未收到消费者的确认，会认为消息消费未被消费而重新发送。

网络不稳定、消费者故障、网络分区、消息重复传递策略、消费者超时设置不当

为什么需要避免重复消费？

业务错误：我本来写的业务逻辑就是只要执行一次

数据重复：数据插入重复，破坏数据唯一性

资源浪费：占用系统资源，降低系统性能

如何避免消息重复消费？

消息去重

通过记录已经消费过的消息，在消息到达时检查它是否已经在记录中存在，从而避免重复处理。

if(!processedMessages.contains(message)){processMessage(message);
                    processedMessages.add(message);}

消息幂等性

分布式锁(消息幂等性)

使用UUID生成唯一Id ，作为messageId

使用了唯一的消息ID来确保同一条消息只会被处理一次。

AMQP.BasicProperties properties =newAMQP.BasicProperties().builder().messageId(UUID.randomUUID().toString())// 唯一标识.build();

if(!isMessageProcessed(messageId)){processMessage(message);saveProcessedMessage(messageId);}

消费者先查询该消息是否已经被处理过，如果没有被处理过，则调用processMessage()方法处理该消息，并使用 saveProcessedMessage()方法保存已经处理过的消息。

//手动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false

在处理完消息后，还需要调用channel.basicAck(envelope.getDeliveryTag(), false)方法确认消息已经被消费。这是因为RabbitMQ是一个消息的投递机制，只有在消费者确认了消息已经被处理后，才会从消息队列中删除该消息。

使用redis实现避免重复消费

生产者

AMQP.BasicProperties properties =newAMQP.BasicProperties().builder().deliveryMode(1)// 指定消息是否需要持久化 1-需要 2-不需要.messageId(UUID.randomUUID().toString())// 唯一标识.build();

消费者

String result = jedis.set(messageId,"0","NX","EX",10);if(result !=null&& result.equalsIgnoreCase("OK")){System.out.println("接收到消息："+newString(body,"UTF-8"));//消费成功 set messageId - 1
                    jedis.set(messageId,"1");
                    channel.basicAck(envelope.getDeliveryTag(),false);}else{//如果1中的setnx失败，获取key对应的value，如果是1，设置ack 如果是0 returnString s = jedis.get(messageId);if("1".equalsIgnoreCase(s)){//消费完了
                        channel.basicAck(envelope.getDeliveryTag(),false);}}

spring-boot

如果存在，设置value为1；如果value是1，ack

事务性消费

消费状态追踪