RabbitMQ如何保证幂等性

一、简介

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

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

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

二、消息全局ID

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

@Configuration
publicclassDirectRabbitConfig{//队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue(){// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。//   return new Queue("TestDirectQueue",true,true,false);//一般设置一下队列的持久化就好,其余两个就是默认falsereturnnewQueue("TestDirectQueue",true);}//Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange(){//  return new DirectExchange("TestDirectExchange",true,true);returnnewDirectExchange("TestDirectExchange",true,false);}//绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect(){returnBindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");}}

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

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

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

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

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

【3】消费者消费消息

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

相关配置代码

@Configuration
publicclassMessageListenerConfig{

    @Resource
    private CachingConnectionFactory connectionFactory;
    @Resource
    private MyAckReceiver myAckReceiver;//消息接收处理类

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(){
        SimpleMessageListenerContainer container =newSimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);// RabbitMQ默认是自动确认，这里改为手动确认消息//设置一个队列
        container.setQueueNames("TestDirectQueue");//如果同时设置多个如下： 前提是队列都是必须已经创建存在的//  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
        container.setMessageListener(myAckReceiver);return container;}}

【4】 消息消费者

@Component
publicclassMyAckReceiverimplementsChannelAwareMessageListener{

    @Override
    publicvoidonMessage(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 =newString(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();}}}

实现效果

](https://img-blog.csdnimg.cn/2ee21b9f8ad245699435fae2b6cbcd04.png)

三、总结

可以看到消费者已经拿到了全局ID，消费完之后成功把队列的消息删除，这样就可以避免消息被重复消费