rabbitmq实现幂等性操作

超、凢脫俗 2023-10-11 17:19 8阅读 0赞

#### 文章目录 ####

*   1.场景描述
 *   *  1.1 场景1
     *  1.2 场景2
 *  2.原理
 *  3.实战开发
 *   *  3.1 建表
     *  3.2 集成mybatis-plus
     *  3.3 集成RabbitMq
     *   *  3.3.1 安装mq
         *  3.3.2 springBoot集成mq
     *  3.4 具体实现
     *   *  3.4.1 mq配置类
         *  3.4.2 生产者
         *  3.4.3 消费者

##  1.场景描述 ##

消息中间件是分布式系统常用的组件，无论是异步化、解耦、削峰等都有广泛的应用价值。我们通常会认为，消息中间件是一个可靠的组件——这里所谓的可靠是指，只要我把消息成功投递到了消息中间件，消息就不会丢失，即消息肯定会至少保证消息能被消费者成功消费一次，这是消息中间件最基本的特性之一，也就是我们常说的“AT LEAST ONCE”，即消息至少会被“成功消费一遍”。

### 1.1 场景1 ###

什么意思呢？举个例子：一个消息M发送到了消息中间件，消息投递到了消费程序A，A接受到了消息，然后进行消费，但在消费到一半的时候程序重启了，这时候这个消息并没有标记为消费成功，这个消息还会继续投递给这个消费者，直到其消费成功了，消息中间件才会停止投递。  
这种情景就会出现消息可能被多次地投递。

### 1.2 场景2 ###

还有一种场景是程序A接受到这个消息M并完成消费逻辑之后，正想通知消息中间件“我已经消费成功了”的时候，程序就重启了，那么对于消息中间件来说，这个消息并没有成功消费过，所以他还会继续投递。这时候对于应用程序A来说，看起来就是这个消息明明消费成功了，但是消息中间件还在重复投递。

以上两个场景对于消息队列来说就是同一个messageId的消息重复投递下来了。

## 2.原理 ##

全局唯一ID + (Redis/数据库)  
原理很简单，我们先看个流程图：  
![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAWk5pbmVTdW4_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center][]  
我们利用消息id来判断消息是否已经消费过，如果该信息被消费过，那么消息表中已经 会有一条数据，由于消费时会先执行插入操作，此时会因为主键冲突无法重复插入，我们就利用这个原理来进行幂等的控制。  
更详细的原理说明和解释可以看一下[《MQ幂等、去重的解决方案》][MQ]  
本文对于原理不在过多赘述，主要是看看怎么在项目中实现的

## 3.实战开发 ##

### 3.1 建表 ###

DROP TABLE IF EXISTS `message_idempotent`;
    CREATE TABLE `message_idempotent` (
      `message_id` varchar(50) NOT NULL COMMENT '消息ID',
      `message_content` varchar(2000) DEFAULT NULL COMMENT '消息内容',
      `status` int DEFAULT '0' COMMENT '消费状态（0-未消费成功;1-消费成功）',
      `retry_times` int DEFAULT '0' COMMENT '重试次数',
      PRIMARY KEY (`message_id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

### 3.2 集成mybatis-plus ###

[《springBoot集成mybatisPlus》][springBoot_mybatisPlus]

### 3.3 集成RabbitMq ###

#### 3.3.1 安装mq ####

推荐使用docker安装rabbitmq，还未安装的可以参考以下信息：

*  [windows下的安装][windows]
 *  [ubuntu下的安装][ubuntu]
 *  [docker安装][docker]

#### 3.3.2 springBoot集成mq ####

*  1.添加依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

*  2.修改application.properties

# 配置rabbbitMq
    spring.rabbitmq.host=192.168.1.2
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=ninesun
    spring.rabbitmq.password=zx12345678
    #开启重试监听，设置重试5次，间隔3s
    # 需要开启手动确认机制，要不然代码中手动确认会无效的
    spring.rabbitmq.listener.simple.acknowledge-mode=manual
    spring.rabbitmq.listener.simple.retry.enabled=true
    spring.rabbitmq.listener.simple.retry.max-attempts=5
    spring.rabbitmq.listener.simple.retry.max-interval=3000ms

### 3.4 具体实现 ###

#### 3.4.1 mq配置类 ####

*  DirectRabbitConfig

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class DirectRabbitConfig {
        
        /**
         * 消息队列，消息通过发送和路由之后最终到达的地方，到达 Queue的消息即进入逻辑上等待消费的状态。每个消息都会被发送到一个或多个队列。
         */
        public static final String QUEUE_NAME = "directQueue";//队列名称
        /**
         * 消息交换机，是消息第一个到达的地方。消息通过它指定的路由规则，分发到不同的消息队列中去。
         */
        public static final String EXCHANGE_NAME = "directExchange";
        /**
         * 路由关键字，Exchange根据这个关键字进行消息投递。
         */
        public static final String ROUTING_KEY = "directRouting";
    
        // 创建一个队列名称为directQueue
        @Bean
        public Queue directQueue() {
        
            /**
             * 在new Queue()的时候是有3个构造方法的，由于参数的不同，创建的结果也不一样，下面说下里面的参数
             *durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
             *exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
             *autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
             * 一般设置一下队列的持久化就好,其余两个就是默认false
             */
            return new Queue(QUEUE_NAME, true);
        }
    
        // 创建一个Direct交换机起名为TestDirectExchange
        @Bean
        public DirectExchange directExchange() {
        
            return new DirectExchange(EXCHANGE_NAME, true, false);
        }
    
        // 将队列和交换机绑定, 并设置用于匹配键：directRouting
        @Bean
        public Binding bindingDirect() {
        
            return BindingBuilder.bind(directQueue()).to(directExchange()).with(ROUTING_KEY);
        }
    }

*  死信队列配置

由于rabbitMq中不直接支持死信队列，需要我们利用插件rabbitmq\_delayed\_messgae\_exchage进行开启  
具体如何开启可以参考[《rabbitMq实现延迟队列-4.4节》][rabbitMq_-4.4]

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.CustomExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class DelayedRabbitMQConfig {
        
        public static final String DELAYED_QUEUE_NAME = "delay.queue.demo.delay.queue";
        public static final String DELAYED_EXCHANGE_NAME = "delay.queue.demo.delay.exchange";
        public static final String DELAYED_ROUTING_KEY = "delay.queue.demo.delay.routingkey";
    
        @Bean
        public Queue immediateQueue() {
        
            return new Queue(DELAYED_QUEUE_NAME);
        }
    
        @Bean
        public CustomExchange customExchange() {
        
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "topic");
            return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
        }
    
        @Bean
        public Binding bindingNotify(@Qualifier("immediateQueue") Queue queue,
                                     @Qualifier("customExchange") CustomExchange customExchange) {
        
            return BindingBuilder.bind(queue).to(customExchange).with(DELAYED_ROUTING_KEY).noargs();
        }
    }

#### 3.4.2 生产者 ####

*  1.消费订单的生产者

import com.example.mq_repeat_idempotent.mq.config.DirectRabbitConfig;
    import com.example.mq_repeat_idempotent.service.IMessageIdempotentService;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.UUID;
    
    @Component
    @Slf4j
    public class Sender_Direct {
        
        @Autowired
        private AmqpTemplate rabbitTemplate;
        @Autowired
        IMessageIdempotentService iMessageIdempotentService;
    
        /**
         * 用于消费订单
         *
         * @param orderId
         */
        public void send2Direct(String orderId) {
        
            log.info("订单Id：" + orderId);
            //创建消费对象，并指定全局唯一ID(这里使用UUID，也可以根据业务规则生成，只要保证全局唯一即可)
            MessageProperties messageProperties = new MessageProperties();
            messageProperties.setMessageId(UUID.randomUUID().toString());
            messageProperties.setContentType("text/plain");
            messageProperties.setContentEncoding("utf-8");
            Message message = new Message(orderId.getBytes(), messageProperties);
            rabbitTemplate.convertAndSend(DirectRabbitConfig.EXCHANGE_NAME, DirectRabbitConfig.ROUTING_KEY, message);
        }
    
    }

*  2.死信队列的生产者

import lombok.extern.slf4j.Slf4j;
    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.stereotype.Component;
    import org.springframework.stereotype.Service;
    
    import java.util.UUID;
    
    import static com.example.mq_repeat_idempotent.mq.config.DelayedRabbitMQConfig.DELAYED_EXCHANGE_NAME;
    import static com.example.mq_repeat_idempotent.mq.config.DelayedRabbitMQConfig.DELAYED_ROUTING_KEY;
    
    @Service
    @Slf4j
    public class DelaySender {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        public void sendDelayMsg(String msg, Integer delayTime) {
        
            log.info("接受到信息为：" + msg + "，延迟：" + delayTime + "s后发送");
            rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, msg, a -> {
        
                a.getMessageProperties().setDelay(delayTime * 1000);
                return a;
            });
        }
    
        public void sendDelayMsg(String msg, String messageId, Integer delayTime) {
        
            log.info("接受到信息为：" + msg + "，延迟：" + delayTime + "s后发送");
            //创建消费对象，并指定全局唯一ID(这里使用UUID，也可以根据业务规则生成，只要保证全局唯一即可)
            MessageProperties messageProperties = new MessageProperties();
            messageProperties.setMessageId(messageId);
            messageProperties.setContentType("text/plain");
            messageProperties.setContentEncoding("utf-8");
            Message message = new Message(msg.getBytes(), messageProperties);
            rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message, a -> {
        
                a.getMessageProperties().setDelay(delayTime * 1000);
                return a;
            });
        }
    }

#### 3.4.3 消费者 ####

import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
    import com.example.mq_repeat_idempotent.mq.config.DelayedRabbitMQConfig;
    import com.example.mq_repeat_idempotent.mq.config.DirectRabbitConfig;
    import com.example.mq_repeat_idempotent.mq.provider.DelaySender;
    import com.example.mq_repeat_idempotent.po.MessageIdempotent;
    import com.example.mq_repeat_idempotent.service.IMessageIdempotentService;
    import com.rabbitmq.client.Channel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import org.springframework.util.StringUtils;
    
    import java.io.IOException;
    import java.util.Date;
    
    import static com.example.mq_repeat_idempotent.mq.config.DelayedRabbitMQConfig.DELAYED_QUEUE_NAME;
    
    
    @Component
    @Slf4j
    public class Receiver_Direct {
        
        @Autowired
        IMessageIdempotentService iMessageIdempotentService;
        @Autowired
        DelaySender delaySender;
        private static final Integer delayTimes = 30;//延时消费时间,单位：秒
    
        @RabbitListener(queues = {
        DirectRabbitConfig.QUEUE_NAME, DELAYED_QUEUE_NAME})
        public void receiveD(Message message, Channel channel) throws IOException {
        
    
            // 获取消息Id
            String messageId = message.getMessageProperties().getMessageId();
            if (StringUtils.isEmpty(messageId)) {
        
                // 开启消息确认机制
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            String msg = new String(message.getBody());//获取消息
            Integer orderId = Integer.valueOf(msg);
            //向数据库插入数据
            MessageIdempotent messageIdempotent = new MessageIdempotent();
            messageIdempotent.setMessageId(messageId);
            messageIdempotent.setMessageContent(msg);
            messageIdempotent.setRetryTimes(0);
            boolean save = true;
            try {
        
                save = iMessageIdempotentService.save(messageIdempotent);//向消息表中插入数据
            } catch (Exception e) {
        
                e.printStackTrace();
                save = false;
            } finally {
        
                if (!save) {
        //说明属于重重复请求
                    //获取消息表的数据来判断是否消费成功
                    messageIdempotent = iMessageIdempotentService.getById(messageId);
                    Integer status = messageIdempotent.getStatus();
                    if (status == 1) {
        //说明已经消费成功，无需重复消费
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手动确认消费成功
                        return;
                    } else {
        //说明未消费成功
                        //重新进行消费
                        if (consumeOrder(orderId, messageId)) {
        //说明消费成功
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手动确认消费成功
                            return;
                        } else {
        //进入死信队列 ，延时消费
                            delaySender.sendDelayMsg(msg, messageId, delayTimes);//30s之后再试
                            return;
                        }
                    }
                } else {
        //说明是第一次进行消费
                    if (consumeOrder(orderId, messageId)) {
        //说明消费成功
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手动确认消费成功
                        return;
                    } else {
        //进入死信队列 ，延时消费
                        delaySender.sendDelayMsg(msg, messageId, 30);//30s之后再试
                        return;
                    }
    
                }
            }
    
        }
    
        private boolean consumeOrder(Integer orderId, String messageId) {
        
            //消费订单业务
            /**
             * 此处用sig模拟订单业务是否执行成功
             * 模拟执行成功就将sig设置为true
             * 模拟执行失败就将sig设置为false
             */
            boolean sig = false;
            if (sig) {
        
                //执行更新操作---注意：如果考虑并发量高的情况下，可以采用悲观锁--selecte ... for update来进行处理
                UpdateWrapper updateWrapper = new UpdateWrapper();
                updateWrapper.eq("message_id", messageId);
                updateWrapper.set("status", 1);//说明已经消费成功
                iMessageIdempotentService.update(updateWrapper);
            }
            return sig;
        }
    }

至此mq的消息重复以及幂等的信息处理就很完美的解决了，当然本文以数据库为例进行实现，感兴趣的可以尝试使用redis来进行实现

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAWk5pbmVTdW4_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center]: https://img-blog.csdnimg.cn/67edf64c49654509b28ff5449fe787f4.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAWk5pbmVTdW4=,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[MQ]: https://blog.csdn.net/zhiyikeji/article/details/124174677?spm=1001.2014.3001.5501
[springBoot_mybatisPlus]: https://blog.csdn.net/zhiyikeji/article/details/116258935
[windows]: https://www.cnblogs.com/saryli/p/9729591.html
[ubuntu]: https://www.cnblogs.com/sui776265233/p/10954324.html
[docker]: https://blog.csdn.net/zhiyikeji/article/details/123474788
[rabbitMq_-4.4]: https://blog.csdn.net/zhiyikeji/article/details/121770609