RocketMQ源码6-producer 容错机制

曾经终败给现在 2023-09-25 17:13 3阅读 0赞

## 1. 前言 ##

本文主要是介绍一下`RocketMQ`消息生产者在发送消息的时候发送失败的问题处理？这里有两个点，一个是关于消息的处理，一个是关于broker的处理，比如说发送消息到broker-a的broker失败了，我们可能下次就不想发送到这个broker-a，这就涉及到一个选择broker的问题，也就是选择MessageQueue的问题。

## 2. 失败重试 ##

其实失败重试我们在介绍`RocketMQ`消息生产者发送消息的时候介绍过了，其实`同步发送与异步发送都会失败重试`的，比如说我发送一个消息，然后超时了，这时候在`MQProducer`层就会进行控制重试，`默认是重试2次`的，加上你发送那次，`一共是发送3次`，如果重试完还是有问题的话，这个时候就会抛出异常了。

我们来看下这一块的代码实现（ `DefaultMQProducerImpl` 类`sendDefaultImpl`方法）：

![format_png][]

这块其实就是用for循环实现的，其实不光`RocketMQ`，分布式远程调用框架`Dubbo`的失败重试也是用for循环实现的。

## 3. 延迟故障 ##

我们都知道，在`RocketMQ`中一个`topic`其实是有多个`MessageQueue`这么一个概念的，然后这些`MessageQueue`可能对应着不同的`broker name`，比如说id是0和1的`MessageQueue`对应的broker name是 `broker-a` ，然后id是2和3的`MessageQueue`对应的`broker name` 是`broker-b`

我们发送消息的时候，其实涉及到发送给哪个`MessageQueue`这么一个问题，当然我们可以在发送消息的时候指定这个`MessageQueue`，如果你不指定的话，`RocketMQ`就会根据`MQFaultStrategy` 这么一个策略类给选择出来一个`MessageQueue`。

我们先来看下是在哪里选择的，其实就是在我们重试的循环中: org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl\#sendDefaultImpl

...
    // 重试发送
    for (; times < timesTotal; times++) {
        String lastBrokerName = null == mq ? null : mq.getBrokerName();
        // todo 选择message queue
        MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
        ...
    复制代码

我们可以看到，它会把`topicPublishInfo` 与 `lastBrokerName` 作为参数传进去，`topicPublishInfo` 里面其实就是那一堆`MessageQueue`， 然后这个`lastBrokerName` 是上次我们选择的那个`broker name` ， 这个接着我们来看下这个`selectOneMessageQueue`实现:

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        // todo
        return this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName);
    }
    复制代码

可以看到它调用了`MQFaultStrategy` 这个类的`selectOneMessageQueue` 方法，我们接着进去:

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        // 发送延迟故障启用，默认为false
        if (this.sendLatencyFaultEnable) {
            try {
                // 获取一个index
                int index = tpInfo.getSendWhichQueue().getAndIncrement();
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    if (pos < 0)
                        pos = 0;
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    // 选取的这个broker是可用的 直接返回
                    if (latencyFaultTolerance.isAvailable(mq.get

[format_png]: https://img-blog.csdnimg.cn/img_convert/aba9fd1e98d525840b5ed7e16490cbc0.webp?x-oss-process=image/format,png