探究 | RocketMQ 发送消息messageQueue选择策略~

素颜马尾好姑娘i 2023-10-15 11:43 56阅读 0赞

# 前言 #

大家好哇，今天跟大家一起了解下RocketMq发送消息时，关于Topic queue的选择策略~

众所周知，Rocketmq中，一个Topic可以拥有多个message queue，并且每个message queue会分布在不同的broker上，那么我们在send一条消息时，这条消息，会被发送到哪个message queue上呢，这就涉及到选择策略了~

--------------------

# 源码探究 #

# 发送消息 #

当我们send一条消息时，最终在sendDefaultImpl里，会通过selectOneMessageQueue来选择一条**消息队列**

java复制代码private SendResult sendDefaultImpl(
      Message msg,
      final CommunicationMode communicationMode,
      final SendCallback sendCallback,
      final long timeout
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    
      // ......
    
      // 获取topic信息
      TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
      if (topicPublishInfo != null && topicPublishInfo.ok()) {
        boolean callTimeout = false;
        MessageQueue mq = null;
        Exception exception = null;
        SendResult sendResult = null;
        int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
        int times = 0;
        String[] brokersSent = new String[timesTotal];
        for (; times < timesTotal; times++) {
          String lastBrokerName = null == mq ? null : mq.getBrokerName();
    
          // todo 选择一个消息队列
          MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
    
          // ......
        }
    
        // ......
      }
    
    }

--------------------

# 队列选择策略 #

java复制代码// org.apache.rocketmq.client.impl.producer.TopicPublishInfo#selectOneMessageQueue
    public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
      if (lastBrokerName == null) {
        // 第一次发送消息，lastBrokerName为null
        return selectOneMessageQueue();
      } else {
        // 消息发送失败重试，此时lastBrokerName就是上一次所选msgQueue所属broker
        for (int i = 0; i < this.messageQueueList.size(); i++) {
          // 自增上次选的队列index
          int index = this.sendWhichQueue.incrementAndGet();
          int pos = Math.abs(index) % this.messageQueueList.size();
          if (pos < 0)
            pos = 0;
          
          MessageQueue mq = this.messageQueueList.get(pos);
          // 避免跟上次选到一样的broker所属队列
          if (!mq.getBrokerName().equals(lastBrokerName)) {
            return mq;
          }
        }
        return selectOneMessageQueue();
      }
    }

上述代码虽然还没看到具体的选择策略，但是我们可以看出RocketMQ发送消息的**容错机制**

即在**消息发送失败进行重试**时，会去选择一个新的队列，并且这个队列所属的broker跟上次的不一样。

下面继续来看看，具体的选择策略是什么样的~

java复制代码// sendWhichQueue队列选择器
    private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
    
    public MessageQueue selectOneMessageQueue() {
      // 自增并获取一个index
      int index = this.sendWhichQueue.incrementAndGet();
      
      // 与messageQueueList.size()取模
      int pos = Math.abs(index) % this.messageQueueList.size();
      if (pos < 0)
        // 如果 < 0，则默认选择第一个queue
        pos = 0;
      
      // 获取messageQueue
      return this.messageQueueList.get(pos);
    }
    
    public class ThreadLocalIndex {
    
      private final ThreadLocal<Integer> threadLocalIndex = new ThreadLocal<Integer>();
    
      private final Random random = new Random();
    
      private final static int POSITIVE_MASK = 0x7FFFFFFF;
    
      public int incrementAndGet() {
        Integer index = this.threadLocalIndex.get();
    
        if (null == index) {
          // 生成一个随机index
          index = Math.abs(random.nextInt());
          this.threadLocalIndex.set(index);
        }
    
        this.threadLocalIndex.set(++index);
    
        return Math.abs(index & POSITIVE_MASK);
      }
    
    }

通过上述代码可见，

1.  对于某个线程来说，第一次来获取queue时是生成的一个**随机数**，然后与queueSize取模，简单理解就是随机选一个队列，并且利用了TheadLocal进行线程隔离。
2.  当消息发送失败时，会进行重新发送，此时会通过ThreadLocal拿到当前线程上一次的index进行自增，即选择上次所选队列的下一个队列，且两个队列所属的broker不同（因为是失败重试，所以要选择新的broker，提高重试的成功率）

--------------------

# 自定义队列选择器 #

通过上面的源码探究，我们了解到了，RocketMq默认send消息时，是随机选择的一个message queue，当消息失败重试时，会选择下一个不同的broker所属的message queue。

但是，在实际开发中，一些特殊场景，比如我们要发送顺序消息，具体逻辑见: ，这里我就不做过多赘述了~

咱们直接上官方案例

java复制代码public class Producer {
      public static void main(String[] args) throws UnsupportedEncodingException {
        try {
          DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
          producer.start();
    
          String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
          for (int i = 0; i < 100; i++) {
            int orderId = i % 10;
            Message msg =
              new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
                          ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            
            // 自定义MessageQueueSelector，即队列选择器
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
              
              // 重写select方法
              @Override
              public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                // arg就是我们传入的orderId
                Integer id = (Integer) arg;
                // 将orderId与queueSize取模，确保同一个订单最终选择的是同一个队列
                int index = id % mqs.size();
                return mqs.get(index);
              }
              // 传入orderId
            }, orderId);
    
            System.out.printf("%s%n", sendResult);
          }
    
          producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
          e.printStackTrace();
        }
      }
    }

**再看一波源码中是怎么调用了**

java复制代码private SendResult sendSelectImpl(
      Message msg,
      MessageQueueSelector selector,
      Object arg,
      final CommunicationMode communicationMode,
      final SendCallback sendCallback, final long timeout
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
     
      // .....省略
    
      TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
      if (topicPublishInfo != null && topicPublishInfo.ok()) {
        MessageQueue mq = null;
        try {
          List<MessageQueue> messageQueueList =
            mQClientFactory.getMQAdminImpl().parsePublishMessageQueues(topicPublishInfo.getMessageQueueList());
          Message userMessage = MessageAccessor.cloneMessage(msg);
          String userTopic = NamespaceUtil.withoutNamespace(userMessage.getTopic(), mQClientFactory.getClientConfig().getNamespace());
          userMessage.setTopic(userTopic);
    
          // 这里调用了selector.select，即调用了我们自定义的select方法来选择MessageQueue
          mq = mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));
        } catch (Throwable e) {
          throw new MQClientException("select message queue threw exception.", e);
        }
    
        // .....省略
      }
    
      validateNameServerSetting();
      throw new MQClientException("No route info for this topic, " + msg.getTopic(), null);
    }

--------------------

# 总结 #

阅读本文，我们可以了解如下知识点

1.  RocketMq发送消息，默认是随机选择的message queue
2.  消息发送失败重试时，会自增index，即选择下一个message queue，且下一个message queue所属broker与上一个不同，以此来提高消息重试的成功率.
3.  在特殊场景下，例如订单顺序消息，我们可自定义message queue的选择器，保证同一个订单号发送到同一个队列中，保证消息的顺序发送。