RocketMq之消息可靠性

深藏阁楼爱情的钟 2023-02-14 15:57 48阅读 0赞

## 1. 消息的发送流程 ##

一条消息从生产到被消费，将会经历三个阶段：

![format_png][]

*  生产阶段，Producer 新建消息，然后通过网络将消息投递给 MQ Broker
 *  存储阶段，消息将会存储在 Broker 端磁盘中
 *  消息阶段， Consumer 将会从 Broker 拉取消息

以上任一阶段都可能会丢失消息，我们只要找到这三个阶段丢失消息原因，采用合理的办法避免丢失，就可以彻底解决消息丢失的问题。

## 2. 生产阶段 ##

生产者（Producer） 通过网络发送消息给 Broker，当 Broker 收到之后，将会返回确认响应信息给 Producer。所以生产者只要接收到返回的确认响应，就代表消息在生产阶段未丢失。

### 发送模式 ###

**可靠同步发送**

*  原理：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式。
 *  场景：此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。
 *  类似推拉的形式 `发送 ->同步返回 ->发送 ->同步返回`

**可靠异步发送**

*  原理：异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。 MQ 的异步发送，需要用户实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务器响应即可返回，进行第二条消息发送。发送方通过回调接口接收服务器响应，并对响应结果进行处理。
 *  场景：异步发送一般用于链路耗时较长，对响应时间较为敏感的业务场景，例如用户视频上传后通知启动转码服务，转码完成后通知推送转码结果等。 耗时比较长的 可以不需要同步返回给用户的

**单向（Oneway）发送**

*  原理：单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。 此方式发送消息的过程耗时非常短，一般在微秒级别。
 *  场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。

RocketMQ 发送消息示例代码如下：

DefaultMQProducer mqProducer=new DefaultMQProducer("test");
    // 设置 nameSpace 地址
    mqProducer.setNamesrvAddr("namesrvAddr");
    mqProducer.start();
    Message msg = new Message("test_topic" /* Topic */,
            "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    );
    // 发送消息到一个Broker
    try {
        SendResult sendResult = mqProducer.send(msg);
    } catch (RemotingException e) {
        e.printStackTrace();
    } catch (MQBrokerException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    复制代码

`send` 方法是一个同步操作，只要这个方法不抛出任何异常，就代表消息已经**发送成功**。

消息发送成功仅代表消息已经到了 Broker 端，Broker 在不同配置下，可能会返回不同响应状态:

*  `SendStatus.SEND_OK`
 *  `SendStatus.FLUSH_DISK_TIMEOUT`
 *  `SendStatus.FLUSH_SLAVE_TIMEOUT`
 *  `SendStatus.SLAVE_NOT_AVAILABLE`

引用官方状态说明：

![image-20200319220927210][]

另外 RocketMQ 还提供异步的发送的方式，适合于链路耗时较长，对响应时间较为敏感的业务场景。

DefaultMQProducer mqProducer = new DefaultMQProducer("test");
    // 设置 nameSpace 地址
    mqProducer.setNamesrvAddr("127.0.0.1:9876");
    mqProducer.setRetryTimesWhenSendFailed(5);
    mqProducer.start();
    Message msg = new Message("test_topic" /* Topic */,
            "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    );
    
    try {
        // 异步发送消息到，主线程不会被阻塞，立刻会返回
        mqProducer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 消息发送成功，
            }
    
            @Override
            public void onException(Throwable e) {
                // 消息发送失败，可以持久化这条数据，后续进行补偿处理
            }
        });
    } catch (RemotingException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

异步发送消息一定要**注意重写**回调方法，在回调方法中检查发送结果。

不管是同步还是异步的方式，都会碰到网络问题导致发送失败的情况。针对这种情况，我们可以设置合理的重试次数，当出现网络问题，可以自动重试。设置方式如下：

// 同步发送消息重试次数，默认为 2
    mqProducer.setRetryTimesWhenSendFailed(3);
    // 异步发送消息重试次数，默认为 2
    mqProducer.setRetryTimesWhenSendAsyncFailed(3);

### 总结 ###

producer消息发送方式虽然有3种，但为了减小丢失消息的可能性尽量采用同步的发送方式，**producer同步等待broker响应消息的发送结果**，利用**同步发送+重试机制+多个master节点**，尽可能减小消息丢失的可能性。

## 3. Broker 存储阶段 ##

默认情况下，消息只要到了 Broker 端，将会优先保存到内存中，然后立刻返回确认响应给生产者。随后 Broker 定期批量的将一组消息从内存异步刷入磁盘。

这种方式减少 I/O 次数，可以取得更好的性能，但是如果发生机器掉电，异常宕机等情况，消息还未及时刷入磁盘，就会出现丢失消息的情况。

若想保证 Broker 端不丢消息，保证消息的可靠性，我们需要将消息保存机制修改为同步刷盘方式，即消息**存储磁盘成功**，才会返回响应。

修改 Broker 端配置如下：

## 默认情况为 ASYNC_FLUSH 
    flushDiskType = SYNC_FLUSH

若 Broker 未在同步刷盘时间内（**默认为 5s**）完成刷盘，将会返回 `SendStatus.FLUSH_DISK_TIMEOUT` 状态给生产者。

**集群部署**

为了保证可用性，Broker 通常采用一主（**master**）多从（**slave**）部署方式。为了保证消息不丢失，消息还需要复制到 slave 节点。

默认方式下，消息写入 **master** 成功，就可以返回确认响应给生产者，接着消息将会异步复制到 **slave** 节点。

> 注：master 配置：flushDiskType = SYNC\_FLUSH

此时若 master 突然**宕机且不可恢复**，那么还未复制到 **slave** 的消息将会丢失。

为了进一步提高消息的可靠性，我们可以采用同步的复制方式，**master** 节点将会同步等待 **slave** 节点复制完成，才会返回确认响应。

异步复制与同步复制区别：

*  Sync Broker：生产者发送的每一条消息都至少同步复制到一个slave后才返回告诉生产者成功，即“同步双写”。
 *  Async Broker：生产者发送的每一条消息只要写入master就返回告诉生产者成功。然后再“异步复制”到slave。

Broker master 节点 同步复制配置如下：

## 默认为 ASYNC_MASTER 
    brokerRole=SYNC_MASTER

如果 **slave** 节点未在指定时间内同步返回响应，生产者将会收到 `SendStatus.FLUSH_SLAVE_TIMEOUT` 返回状态。

### 总结 ###

**在broker端，消息丢失的可能性主要在于刷盘策略和同步机制。**  
RocketMQ默认broker的刷盘策略为异步刷盘，如果有主从，同步策略也默认的是异步同步，这样子可以提高broker处理消息的效率，但是会有丢失的可能性。因此可以通过**同步刷盘策略+同步slave策略+主从的方式**解决丢失消息的可能。

结合生产阶段与存储阶段，若需要**严格保证消息不丢失**，broker 需要采用如下配置：

## master 节点配置
    flushDiskType = SYNC_FLUSH
    brokerRole=SYNC_MASTER
    
    ## slave 节点配置
    brokerRole=slave
    flushDiskType = SYNC_FLUSH

同时这个过程我们还需要生产者配合，判断返回状态是否是 `SendStatus.SEND_OK`。若是其他状态，就需要考虑补偿重试。

虽然上述配置提高消息的高可靠性，但是会**降低性能**，生产实践中需要综合选择。

## 4. 消费阶段 ##

从producer投递消息到broker，即使前面这些过程保证了消息正常持久化，但如果consumer消费消息没有消费到也不能理解为消息绝对的可靠。因此RockerMQ默认提供了At least Once机制保证消息可靠消费。

**何为At least Once？**

Consumer先pull 消息到本地，消费完成后，才向服务器返回ack。

通常消费消息的ack机制一般分为两种思路：

> 1、先提交后消费；
> 
> 2、**先消费，消费成功后再提交**；

思路一可以解决重复消费的问题但是会丢失消息，因此Rocket默认实现的是思路二，由各自consumer业务方保证幂等来解决重复消费问题。

消费者从 broker 拉取消息，然后执行相应的业务逻辑。一旦执行成功，将会返回 `ConsumeConcurrentlyStatus.CONSUME_SUCCESS` 状态给 Broker。

如果 Broker 未收到消费确认响应或收到其他状态，消费者下次还会再次拉取到该条消息，进行重试。这样的方式有效避免了消费者消费过程发生异常，或者消息在网络传输中丢失的情况。

消息消费的代码如下：

// 实例化消费者
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer");
    
    // 设置NameServer的地址
    consumer.setNamesrvAddr("namesrvAddr");
    
    // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
    consumer.subscribe("test_topic", "*");
    // 注册回调实现类来处理从broker拉取回来的消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            // 执行业务逻辑
            // 标记该消息已经被成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    // 启动消费者实例
    consumer.start();

以上消费消息过程的，我们需要**注意返回消息状态**。只有当业务逻辑真正执行成功，我们才能返回 `ConsumeConcurrentlyStatus.CONSUME_SUCCESS`。否则我们需要返回 `ConsumeConcurrentlyStatus.RECONSUME_LATER`，稍后再重试。

## 5. 总结 ##

最后我们还可以说出我们的思考，虽然提高消息可靠性，但是可能导致消息重发，重复消费。所以对于消费客户端，需要注意保证**幂等性**。

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91c2VyLWdvbGQtY2RuLnhpdHUuaW8vMjAyMC8zLzI1LzE3MTBmNWVkODE3YTlmZTI?x-oss-process=image/format,png
[image-20200319220927210]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91c2VyLWdvbGQtY2RuLnhpdHUuaW8vMjAyMC8zLzI1LzE3MTBmNWVlMWQwMTBkMTQ?x-oss-process=image/format,png