《RabbitMQ系列教程-第十三章-消息的幂等性》

绝地灬酷狼 2023-01-02 14:24 102阅读 0赞

# 教程说明 #

*  本系列教程目录大纲：[《RabbitMQ系列教程-目录大纲》][RabbitMQ_-]

--------------------

*  本系列教程配套代码：[https://gitee.com/Horizon1024/rabbitmt.git][https_gitee.com_Horizon1024_rabbitmt.git]（码云地址）

--------------------

# 消息的幂等性 #

## 13.1 幂等简介 ##

幂等性：用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用，例如用户在下订单时，由于网络卡顿等原因多次点击下单按钮，但最终订单只生成了一个，并不会多次下订单；

消息幂等性：即消息无论被发送多少次，最终只会消费一次。保证消息不会重复被消费多次；

### 13.1.1 方案 ###

实现消息的幂等性通常有两种方案来解决：全局消息ID、Redis标识，两种方案从原理上讲都是一样的；

*  全局消息ID：发送消息时给消息分配一个全局ID，每次消费消息时判断该ID是否存在过，如果存在过则已经消费过，如果不存在则说明是第一次消费；
 *  Redis标识：发送消息时，给消息分配一个全局的唯一ID，消费消息时，将id与消息以K-V的形式写入Redis`<id,meessage>`，如果能写入成功代表第一次消费消息，如果写入不成功代表消费已经被消费过了

### 13.1.2 解决消息幂等性 ###

搭建SpringBoot整合RabbitMQ工程

> **我们采用的是Redis的setnx命令来解决消息幂等性问题；**

#### application.yml： ####

spring:
      rabbitmq:
        host: 192.168.133.151
        port: 5672
        username: guest
        password: guest
        virtual-host: /
      redis:
        host: 127.0.0.1

#### 启动类： ####

package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.QueueBuilder;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
    import org.springframework.data.redis.serializer.StringRedisSerializer;
    
    @SpringBootApplication
    public class Application { 
        public static void main(String[] args) { 
            SpringApplication.run(Application.class);
        }
    
        @Bean
        public RedisTemplate redisTemplate(RedisTemplate redisTemplate){ 
    
            // value可见
            redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
    
            // key可见
            redisTemplate.setKeySerializer(new StringRedisSerializer());
    
            return redisTemplate;
        }
    
        // 定义测试队列
        @Bean
        public Queue testQueue(){ 
            return QueueBuilder.durable("test_queue").build();
        }
    }

#### 监听器： ####

package com.lscl.rabbitmq.listener;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.stereotype.Component;
    
    @Component
    public class TestListener { 
    
        @Autowired
        private RedisTemplate redisTemplate;
    
        @RabbitListener(queues = "test_queue")
        public void test_queue_confirm(Message message) { 
    
            String messageId = message.getMessageProperties().getMessageId();
    
            if (null == messageId) { 
    
                System.out.println("消息id为null!");
                return;
            }
    
            if (redisTemplate.opsForValue().setIfAbsent(messageId, new String(message.getBody()))) { 
    
                // 代表第一次消费消息
                System.out.println("消息消费成功: " + new String(message.getBody()));
            } else { 
                System.out.println("消息已经被消费过了!");
            }
        }
    }

#### 消息发送端： ####

package com.lscl.rabbitmq.controller;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    public class TestController { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @GetMapping("/send/{messageId}")
        public String addOrders(@PathVariable String messageId) { 
    
            MessageProperties prop = new MessageProperties();
            prop.setMessageId(messageId);
            Message message = new Message("测试幂等性".getBytes(), prop);
    
            // 发送消息
            rabbitTemplate.convertAndSend("test_queue", message);
    
            return messageId;
        }
    }

[RabbitMQ_-]: https://blog.csdn.net/Bb15070047748/article/details/112194912
[https_gitee.com_Horizon1024_rabbitmt.git]: https://gitee.com/Horizon1024/rabbitmt.git