RabbitMQ如何保证幂等性-Toy模板网

这篇具有很好参考价值的文章主要介绍了RabbitMQ如何保证幂等性。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、简介

幂等性是分布式中比较重要的一个概念，是指在多作业操作时候避免造成重复影响，其实就是保证同一个消息不被消费者重复消费两次，但是可能存在网络波动等问题，生产者无法接受消费者发送的ack信息，因此这条消息将会被重复发送给其他消费者进行消费，实际上这条消息已经被消费过了，这就是重复消费的问题。

如何避免重复消费的问题
1.消息全局唯一ID
2.通过redis中的setnx命令，给消息分配一个全局ID，当消费了这个消息的时候写入redis

----下面通过第一种方式示例来避免重复消费

二、消息全局ID

【1】将队列和交换机绑定

@Configuration
public class DirectRabbitConfig {

    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);

        //一般设置一下队列的持久化就好,其余两个就是默认false

        return new Queue("TestDirectQueue",true);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        //  return new DirectExchange("TestDirectExchange",true,true);
        return new DirectExchange("TestDirectExchange",true,false);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
}

【2】写一个接口来发送消息

@RestController
public class SendMessageController {
    @Resource
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法

    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        MessageProperties messageProperties = new MessageProperties();

        messageProperties.setMessageId(UUID.randomUUID().toString());
        messageProperties.setContentType("text/plain");
        messageProperties.setContentEncoding("utf-8");
        Message message = new Message("hello,message idempotent!".getBytes(), messageProperties);

        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting",message);
        return "ok";
    }
}

【3】消费者消费消息

注：这里是手动确认模式通过实现ChannelAwareMessageListener方法的方式重写了onMessage方法，可以实现ack

【4】消息消费者

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){

                /*
                 * 消费消息之前，根据message中设置的全局ID查看消息是否已经被消费过
                 */
                String messageId = message.getMessageProperties().getMessageId();

                if (StringUtils.isBlank(messageId)){
                    System.out.println("获取ID为空");
                    return;
                }
                String str = "----数据库业务操作-------";
                if (StringUtils.isNotBlank(str)){
                    // 消费当前消息，并获取message内容
                    String msg = new String(message.getBody(), StandardCharsets.UTF_8);
                    System.out.println("---------生产者发送消息内容--------------messageId："+ messageId + ",内容："+ msg +"");

                    // 消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息（成功消费，消息从队列中删除 ）
                    channel.basicAck(deliveryTag, false);
                }else {

                    // ack返回false，requeue-true并重新回到队列
                    channel.basicNack(deliveryTag, false, true);
                }
            }
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
}