RabbitMQ避免重复消费

本文介绍: 幂等性是指无论操作执行多少次，都是得到相同的结果，而不会产生其他副作用。在r abb i tMQ中同一条消息在MQ中被消费多次。

在Jav a中，可以使用消息队列来实现消息的异步处理，其中常用的消息队列有 Rabb i tMQ、Ac ti v eMQ、Kaf k a 等。

什么是幂等性？

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

在r abb i tMQ中

如何避免消息重复消费？

消息去重

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

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

消息幂等性

分布式锁(消息幂等性)

使用UUID生成唯一Id ，作为mes sageId

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

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

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

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

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

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

使用redis 实现避免重复消费

生产者

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

消费者

 String result = jedis.set(messageId, "0", "NX", "EX", 10);
  
 if (result != null &amp;&amp; result.equalsIgnoreCase("OK")){
                    System.out.println("接收到消息："+ new String(body,"UTF-8"));

                    //消费成功 set messageId - 1
                    jedis.set(messageId,"1");
                    channel.basicAck(envelope.getDeliveryTag(),false);

                }else {
                    //如果1中的setnx失败，获取key对应的value，如果是1，设置ack 如果是0 return

                    String s = jedis.get(messageId);
                    if ("1".equalsIgnoreCase(s)){
                        //消费完了
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }