RocketMq消息0丢失方案

左手的ㄟ右手 2022-12-27 01:45 238阅读 0赞

### 文章目录 ###

*   *  消息会在哪里丢失
     *  生产者100%投递消息
     *  RocketMq100%不丢失消息
     *  Consumer 100% 消费消息

## 消息会在哪里丢失 ##

在想要解决RocketMq消息丢失问题我们需要先考虑消息会在哪里丢失。我们从消息链路来依次排查

1.  生产者发送消息消息丢失
2.  RocketMq本身导致消息丢失
3.  消费者消费失败导致消息丢失

在知道消息会在这三个地方丢失我们就好基于这三个地方来作出相应的解决方案

## 生产者100%投递消息 ##

这里我们以订单支付为例，需要发送订单支付成功到优惠券系统  
首先我们需要保证生产者消息能够100%投递到MQ中，这样才能保证生产者的消息不丢失，具体如何做到呢？主要是是基于RocketMq的事务来保证的，具体步骤如下

1.  首先我们在发送消息到RocketMq中时，我们先会发送一条事务消息half到mq，试探RocketMq是否存活，没有挂掉，这条事务消息由于没有提交事务，所以在消费者端是看不到这条消息的。  
    这里就会有两种情况：
    
    1.  发送half消息成功，继续执行相关业务代码
    2.  发送消息失败，回滚相关业务,这里的相关代码大致如下

// 生产者 
    TransactionMQProducer producer = new TransactionMQProducer("test_product_group"); // 组名 
    Message message = new Message("test_topic", "test_tag", "test_key", "支付成功消息".getBytes(RemotingHelper.DEFAULT_CHARSET));  
    try {   
      // 发送事务消息 
      SendResult sendResult = producer.sendMessageInTransaction(message, null);  
    } catch (MQClientException e) {   
      // half消息发送事务消息失败 
      //订单系统执行回滚， 支付退款，订单关闭，还原库存等 
    }

> 这里消息发送失败又分两种
> 
> 1.  发送half消息真的失败
> 2.  RocketMq接受到half消息，但因为网络原因导致没有没有返回成功结果给生产者  
>     不过是第一种还是第二种失败我们都会执行我们的回滚操作，但是第二种失败导致在half的消息一直在RocketMq但是得不到进一步的commit或rollback，这时RocketMq的 half默认在10分钟没有收到commit或rollback则会回调我们的监听器函数checkLocalTransaction()，代码如下

class TransactionListenerImpl implements TransactionListener{   
      
      // 如果half消息发送成功则调用这个函数 
      @Override  
     public LocalTransactionState executeLocalTransaction(Message message, Object o) {   
      
      // 执行本地订单相关事务，根据订单相关事务执行成功或失败选择RocketMq中half commit or rollback 
      try {   
      
      //执行成功返回 commit return LocalTransactionState.COMMIT_MESSAGE; 
     } catch (Exception e) {   
      
      log.error("发生异常", e);  
      // 订单事务异常 回滚RocketMq half 
      return LocalTransactionState.ROLLBACK_MESSAGE;  
      
     } }  
      // 如果各种原因没有返回commit或rollback则执行这个函数 
      @Override  
     public LocalTransactionState checkLocalTransaction(MessageExt msg) {   
      // 查询本都事务是否都执行完成(订单状态是否支付等) 
      Integer status = orderImpl.get(msg.getTransactionId());  
      if (null != status) {   
      switch (status) {   
      case 0 :  
      return LocalTransactionState.UNKNOW;  
      case 1 :  
      return LocalTransactionState.COMMIT_MESSAGE;  
      case 2 :  
      return LocalTransactionState.ROLLBACK_MESSAGE;  
     } }  return LocalTransactionState.COMMIT_MESSAGE;  
     }}

> 这里因为订单已经回滚，所以我们根据订单状态也会回滚RocketMq中的half

1.  消息发送half成功后，RocketMq就会回调函数executeLocalTransaction()，我们在executeLocalTransaction中就可以执行相关的业务代码

此时这里执行业务代码也会有如下情况  
1. 业务代码执行成功，直接向RocketMq提交commit，如果在commit中发送了异常，则会在catch中执行RocketMq的rollback。如果此时系统宕机或因网络原因导致发送rollback到RocketMq失败，RocketMq在长时间得不到rollback or commit，也会调用我们的函数checkLocalTransaction然后检测订单执行相应的rollback操作

到此就保证了生产则100%投递成功消息到RocketMq，接下来我们就需要保证RocketMq100%不丢失消息

## RocketMq100%不丢失消息 ##

首先我们需要了解的是RocketMq的大致架构是分为NameServer和Broker  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70]

首先是NameServer，虽然NameServer挂掉prodcut还是能够通过本地缓存发送消息，但是如果prodcut重启或者是集群状态有变化，还是不能继续发送消息，所以为了高可用保证我们需要对NameServer进行集群部署，所以目前的架构就演化为如下情况  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 1]

集群部署保证了NameServer的高可用，但是消息存储的核心是Broker，我们需要保证Broker消息的不丢失，首先是消息刷盘策略，RocketMq为了保证高吞吐量默认使用的是异步刷盘策略，即消息投递到RocketMq的os cache即表示消息投递成功，然后通过后台线程定时刷新os cache 到磁盘，由于消息是存放在os cache中，所以也会有消息丢失风险，如果在os cache还没来得及刷入磁盘，然后Broker挂掉，也会导致消息丢失，所以这里我们为了保证消息不丢失，需要将RocketMq默认的异步刷盘策略改为同步刷盘策略(将Broker配置文件的flushDiskType设置为SYNC\_FLUSH，默认为ASYNC\_FLUSH)。  
消息如果刷入到磁盘就能保证消息一定不会丢失吗？如这里单节点的Broker机器的磁盘损坏了，我们的消息还是会丢失，所以我们的Broker就需要主备部署，所以架构就演化为如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 2]

我们将Broker基于DLedger几区和Raft协议部署的主从同步架构就可以保证即使主Broker 磁盘损坏也可以保证其他Slave Broker仍然有消息

自此MQ消息0丢失也保证了，最后需要保证的就是Consumer100%消费消息

## Consumer 100% 消费消息 ##

Consumer 100%消费消息的保证是基于手动offset + 自动故障转义保证的

//消费者组
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer_group");
            // 从NameServer获取路由信息
            consumer.setNamespace("localhost:9876");
            //订阅topic
            consumer.fetchSubscribeMessageQueues("test_topic");
    
            new Thread( () -> { 
                // 获取消息回调
                consumer.registerMessageListener(new MessageListenerConcurrently() { 
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                    ConsumeConcurrentlyContext context) { 
                        
                        
                        //业务代码 发送优惠券 通知发货等
                        
                        // 返回消费消息成功
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                });
                // 启动消费者实例
                try { 
                    consumer.start();
                } catch (MQClientException e) { 
                    e.printStackTrace();
                }
                log.info("消费者启动");
                //当前线程 死循环 一直消费数据
                while (true) { 
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }).start();

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70]: /images/20221120/4cfca01f8cb546a08c1f81072305e474.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 1]: /images/20221120/9d63378d6e8246cbb3c93c9aa0f27cb0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 2]: /images/20221120/46951fb277834cacb59c2c46f1771442.png