手把手带你实现 rabbitmq 保证消息不丢失 MQ消息稳定性

梦里梦外; 2022-10-25 05:52 202阅读 0赞

### 文章目录 ###

*   *  MQ运行图
     *  哪些步骤会发生消息丢失
     *  解决方案
     *   *  MQ宕机之后消息丢失
         *  生产者到MQ之间丢失
         *   *  两种机制
             *  三种confirm模式
     *  消费者没处理完消息，自己直接宕机了
     *   *  未投递到对应的Exchange中导致消息丢失
     *  实践
     *   *  使用springboot做项目
         *   *  1.导入jar包
             *  2. 添加配置
             *  3. 添加回调方法
             *  4. 在配置文件中为RabbitTemplate 设置回调方法
             *  5.生产者发送消息
             *  6.消费者接收消息
             *  7.定时任务定时扫描数据库，进行丢失消息的重发

## MQ运行图 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]

## 哪些步骤会发生消息丢失 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]

1.  生产者到MQ之间丢失，网络抖动通信异常。
2.  MQ中的消息还没有完成异步存盘，MQ宕机，再重启消息丢失。
3.  消费者获取到消息之后，没有来得及处理完毕，自己直接宕机了。
4.  属于操作不当，消费者没有将消息投递到对应的Exchange中导致消息丢失

## 解决方案 ##

### MQ宕机之后消息丢失 ###

**开启消息持久化**  
MQ将消息持久化到磁盘中通常情况是，等到有几千条消息的时候，会**一次性的刷盘到磁盘上面**。消息在MQ内存中还没有刷盘到磁盘就宕机了，消息依旧还是会丢失。需要下面的confirm机制，confirm同时也保证生产者感知消息投递的成功与失败。

### 生产者到MQ之间丢失 ###

RabbitMQ引入了事务机制和发送方确认机制。

#### 两种机制 ####

1.  事务机制  
    过于耗费性能所以**一般不用**，是同步操作，一条消息发送之后会使发送端阻塞，以等待RabbitMQ的回应，之后继续发送下一条消息，生产者生产消息的吞吐量和性能都会大大降低。
2.  发送方确认机制  
    消息发送到MQ之后，MQ会回一个确认收到的消息给我们。将信道设置为confirm模式，在该信道上面发布的消息都会被指派一个唯一的ID，一旦消息被投递到所有匹配的队列之后，RabbitMQ就会发送一个Ack给生产者（包含消息的唯一deliveryTag和multiple参数）**（如果开启了消息与信道持久化，会等到消息存储到磁盘再返回）**

#### 三种confirm模式 ####

1.  串行confirm模式：  
    每发送一条消息后，调用waitForConfirms()方法，等待MQ的响应，这个时候如果MQ端返回false或者在超时时间内未返回，客户端进行消息重传。
2.  批量confirm模式  
    producer每发送一批消息后，调用waitForConfirms()方法，等待MQ的响应，决定是否将一批消息的重发。
3.  异步confirm模式  
    提供一个回调方法，MQ确认了一条或者多条消息后会回调这个方法。

## 消费者没处理完消息，自己直接宕机了 ##

消费者获取到消息之后，没有来得及处理完毕，自己直接宕机了。**消息者默认采用自动ack**（接收到消息就ack），此时消息就丢了。可以采用手动ack机制来解决这个问题，消费端处理完逻辑之后手动通知MQ，如果在消费者拿到消息，没来得及处理的情况下自己挂了，此时MQ集群会自动感知到，它就会重发消息给其他的消费者服务实例。

### 未投递到对应的Exchange中导致消息丢失 ###

相应交换机未找到，消息没有正确投递。设定一个备份交换机，当未有交换机匹配时会将消息投递到此处。

## 实践 ##

实现原理图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 3]

RabbitMq的下载安装以及设置这里就不说了。  
模拟一个下单的流程，下订单，系统扣减库存，再通知积分系统进行积分累加。

### 使用springboot做项目 ###

几种模式的springboot整合rabbitmq：[springboot整合rabbitmq][springboot_rabbitmq]

#### 1.导入jar包 ####

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

#### 2. 添加配置 ####

spring:
      #项目名
      application:
        name: rabbitmq-provider
      #配置rabbitMq 服务器
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        # 生产者开启发送确认
        publisher-confirm-type: correlated
        listener:
          type: simple
          simple:
            #消费端采用手动应答
            acknowledge-mode: manual
            #限制mq推给消费者的消息数量为一条数据
            prefetch: 1

#### 3. 添加回调方法 ####

@Component
    public class RabbitCallback implements RabbitTemplate.ConfirmCallback{ 
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) { 
            if (ack) { 
                System.out.println("消息发送成功:"+correlationData);
                //将数据库中消息id为correlationData的消息进行状态转换为已发送
                updateStatus(correlationData.getId());
            } else { 
                System.out.println("消息发送失败:"+cause);
               //根据策略考虑是否进行重新发送
               dosomeing();
            }
        }
    }

#### 4. 在配置文件中为RabbitTemplate 设置回调方法 ####

@Configuration
    public class RabbitMQConfig { 
        @Autowired
        private CachingConnectionFactory connectionFactory;
        @Autowired
        private RabbitCallback callback;
        /* * 需要对rabbitTemplate设置ConfirmCallback对象， * @return */
        @Bean
        public RabbitTemplate rabbitTemplate() { 
            RabbitTemplate template = new RabbitTemplate(connectionFactory);
            template.setConfirmCallback(callback);
            return template;
        }
    }

#### 5.生产者发送消息 ####

发送消息时，**需要将消息id new CorrelationData(UUID.randomUUID().toString())一并发送。**

@RestController
    public class ProducerController { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
        @Autowired
        private StockDao stockDao;
        @Autowired
        private MsgDao msgDao;
        
        @GetMapping(value = "/hello")
        public String addOrder(){ 
            //扣减库存1
            stock.deduction(1)
            //将消息进行落库
            Msg msg = msgDao.Insert();
            //发送消息
            rabbitTemplate.convertAndSend("route",key, "Msg:"+key,new CorrelationData(UUID.randomUUID().toString()));
            return "success";
        }
    }

#### 6.消费者接收消息 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 4]

消费端配置

#消费端采用手动应答
            acknowledge-mode: manual
            #限制mq推给消费者的消息数量为一条数据
            prefetch: 1

消息消费之前需要根据业务场景判断是否可以重复消费，如果不能需要加入幂等性设计。不重复消费消息。

@Component
    @RabbitListener(...)//根据需要配置
    public class WorkConsumer2 { 
    
        @RabbitHandler
        public void consumer(Message message, Channel channel) throws IOException { 
              //没有消费过进行消费 否则丢弃
             if(!consumed(message.getMessageProperties().getDeliveryTag())) { 
                 //消费消息
                 //确认
                 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
             }
        }
    }

#### 7.定时任务定时扫描数据库，进行丢失消息的重发 ####

定时检查是否有长时间状态未变更为已发送状态的消息，进行消息重发

@Component
    public class Task { 
        
        @Scheduled(cron="30 /5 * * * ? ")
        public void check(){ 
            //消息重发
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]: /images/20221024/c53ad738975041a2b4b6a60bd05588fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]: /images/20221024/fe50704f15d6459a9f2545a111587e75.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]: /images/20221024/ff9d937b2d8444feb296393f305fdfdc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 3]: /images/20221024/4811465a1cb94099832570991704cd7a.png
[springboot_rabbitmq]: https://blog.csdn.net/wsdfym/article/details/113483474
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 4]: /images/20221024/f2d78a3fe394419c978f6e2ba795167d.png