RocketMQ中MessageQueue的选择原理

冷不防 2024-03-24 16:49 56阅读 0赞

有两种情况，一种是正常发送，一种是发送失败后的重试发送

#### 第一种：正常发送消息 ####

这是正常发送消息的内部逻辑，

1. 通过调用 sendWhichQueue.incrementAndGet() 方法获得一个 index 值

2. 将 index 与队列数取余 得到处理后的 pos 下标值

3. 根据 pos 下标取得应该发送的队列

public MessageQueue selectOneMessageQueue() {
      // 拿到 sendWhichQueue
      int index = this.sendWhichQueue.incrementAndGet();
      // 对 messageQueue 列表长度取余
      int pos = Math.abs(index) % this.messageQueueList.size();
      // 计算出来的 pos 如果小于 0, 就给个兜底 = 0
      if (pos < 0)
        pos = 0;
      return this.messageQueueList.get(pos);
    }

这是 sendWhichQueue.incrementAndGet() 方法的逻辑

public int incrementAndGet() {
      Integer index = this.threadLocalIndex.get();
      if (null == index) {
        index = Math.abs(random.nextInt());
        this.threadLocalIndex.set(index);
      }
    
      this.threadLocalIndex.set(++index);
      return Math.abs(index & POSITIVE_MASK);
    }

此方法先判断index是否存在

—— 存在：++index并返回

——不存在：random随机造一个index保存起来

**思考：为什么要 ++index 呢？**

假设第一次随机到的是 index=11，队列数为5，那么采用取余计算 11%5 为1，所以第一次会选择下标为1的队列，之后每次调用如果让 index 自增，那么就是 12%5=2 ，以此类推最终选择的队列顺序为：

1，2，3，4，0，1，2，3，4，0

**没错，实现了线性轮询的负载均衡算法**

它可以将流量均匀地分发给不同的 MessageQueue，而 MessageQueue 分布在不同的 Broker 上，这样也达到了对最终 Message 存储的负载均衡，避免造成数据倾斜。

#### 第二种：在发送模式为SYNC时，重试发送消息 ####

目的是为了避开上一次发送失败的Broker，尽量保证发送成功。

lastBrokerName代表上次选择的Broker

1. 首先判断 lastBrokerName 是否为 null ， lastBrokerName 只有在重试发送消息时才会有值。所以如果为 null 就代表不是重试发送消息，直接采用第一种方式选择队列。

2. 如果 lastBrokerName 为 null ，代表是重试发送消息，进入重试逻辑

3. 取出 index 并取余计算 pos

4. 遍历所有队列，寻找与上一次选择不一样的队列进行发送消息。

public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
        if (lastBrokerName == null) {
            return selectOneMessageQueue();
        } else {
            for (int i = 0; i < this.messageQueueList.size(); i++) {
                int index = this.sendWhichQueue.incrementAndGet();
                int pos = Math.abs(index) % this.messageQueueList.size();
                if (pos < 0)
                    pos = 0;
                MessageQueue mq = this.messageQueueList.get(pos);
                if (!mq.getBrokerName().equals(lastBrokerName)) {
                    return mq;
                }
            }
            return selectOneMessageQueue();
        }
    }