5张图带你理解 RocketMQ 顺序消息实现机制

2022-06-29 程序那点事

加入收藏

作者：君哥聊技术

来源：
https://mp.weixin.qq.com/s/n9QlZ73SQyCGIyPLvHMy0A

大家好，我是君哥。今天聊一聊 RocketMQ 的顺序消息实现机制。

在有些场景下，使用 MQ 需要保证消息的顺序性，比如在电商系统中，用户提交订单、支付订单、订单出库这 3 个消息应该保证顺序性，如下图：

对于 RocketMQ 来说，主要是通过 Producer 和 Consumer 来保证消息顺序的。

1 Producer

下面代码是 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));
   SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
     Integer id = (Integer) arg;
     int index = id % mqs.size();
     return mqs.get(index);
    }
   }, orderId);

   System.out.printf("%s%n", sendResult);
  }

  producer.shutdown();
 } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
  e.printStackTrace();
 }
}

跟发送并发消息不一样的是，发送消息时传入了 MessageQueueSelector，这里可以指定消息发送到固定的 MessageQueue。

注意：上面的代码把 orderId 相同的消息都会发送到同一个 MessageQueue，这样同一个 orderId 的消息是有序的，这也叫做局部有序。对应的另一种是全局有序，这需要把所有的消息都发到同一个 MessageQueue。

下面再来看一下发送的代码：

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);

   mq = mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));
  } catch (Throwable e) {
   throw new MQClientException("select message queue threw exception.", e);
  }

  //省略部分逻辑
  if (mq != null) {
   return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout - costTime);
  } else {
   throw new MQClientException("select message queue return null.", null);
  }
 }
    //省略部分逻辑
}

可以看到，在发送的时候，使用 MessageQueueSelector 选择一个 MessageQueue，然后发送消息到这个 MessageQueue。对于并发消息，这里不传 MessageQueueSelector，如果发送方法没有指定 MessageQueue，就会按照默认的策略选择一个。

2 Consumer

以 RocketMQ 推模式为例，消费者会注册一个监听器，进行消息的拉取和消费处理，下面的 UML 类图显示了调用关系：

上图中包含了对顺序消息和对并发消息的处理。其中 MessageListenerOrderly 和 ConsumeMessageOrderlyService 对顺序消息进行处理。跟并发消息不一样的是，顺序消息定义了一个 MessageQueueLock 类，这个类保存了每个 MessageQueue 对应的锁，代码如下：

private ConcurrentMap<MessageQueue, Object> mqLockTable = new ConcurrentHashMap<MessageQueue, Object>();

下面代码是顺序消费的官方示例：

public static void main(String[] args) throws MQClientException {
 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");

 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

 consumer.subscribe("TopicTest", "TagA || TagC || TagD");

 consumer.registerMessageListener(new MessageListenerOrderly() {
  AtomicLong consumeTimes = new AtomicLong(0);

  @Override
  public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
   context.setAutoCommit(true);
   System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
   this.consumeTimes.incrementAndGet();
   if ((this.consumeTimes.get() % 2) == 0) {
    return ConsumeOrderlyStatus.SUCCESS;
   } else if ((this.consumeTimes.get() % 5) == 0) {
    context.setSuspendCurrentQueueTimeMillis(3000);
    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
   }

   return ConsumeOrderlyStatus.SUCCESS;
  }
 });

 consumer.start();
 System.out.printf("Consumer Started.%n");
}

下面看一下顺序消息的消费端处理逻辑。

2.1 注册监听

上面的代码定义了顺序消息监听器 MessageListenerOrderly，并且注册到 DefaultMQPushConsumer，这个注册同时也注册到了 DefaultMQPushConsumerImpl。

2.2 PushConsumer 初始化

在 DefaultMQPushConsumerImpl 类初始化的时候，会判断注册的 MessageListener 是不是 MessageListenerOrderly，如果是，就把 consumeOrderly 变量设置为 true，以此来标记是顺序消息拉取还是并发消息拉取。然后把 ConsumeMessageService 初始化为 ConsumeMessageOrderlyService。

2.3 锁定 mq

要保证消息的顺序性，就需要保证同一个 MessageQueue 只能被同一个 Consumer 消费。

ConsumeMessageOrderlyService 初始化的时候，会启动一个定时任务，周期性（默认 20s）地向 Broker 发送锁定消息（请求类型 LOCK_BATCH_MQ），Broker 收到后，就会把 MessageQueue、group 和 clientId 进行绑定，这样其他客户端就不能从这个 MessageQueue 拉取消息。

注意：Broker 锁定是有过期时间的，默认 60s，可以配置，锁定过期后，有可能被其他 Consumer 进行消费。

Broker 端锁结构如下图：

2.4 拉取消息

消费者启动时，启动消费拉取线程 PullMessageService，里面死循环不停地从 Broker 拉取消息。这里调用了 DefaultMQPushConsumerImpl 类的 pullMessage 方法。这里拉取消息的逻辑跟并发消息的逻辑是一样的。

拉取到消息后，调用 PullCallback 的 onSuccess 方法处理结果，这里调用了 ConsumeMessageOrderlyService 的 submitConsumeRequest 方法，里面用线程池提交了 ConsumeRequest 线程。

PullCallback pullCallback = new PullCallback() {
 @Override
 public void onSuccess(PullResult pullResult) {
  if (pullResult != null) {
   pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrApper.processPullResult(pullRequest.getMessageQueue(), pullResult,
    subscriptionData);

   switch (pullResult.getPullStatus()) {
    case FOUND:
     //省略
     if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
      DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
     } else {
      //省略
      boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
      DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
       pullResult.getMsgFoundList(),
       processQueue,
       pullRequest.getMessageQueue(),
       dispatchToConsume);
                        //省略
     }
     //省略
     break;
    //省略
   }
  }
 }
    //省略
};

上面拉取到消息后，先把消息放到了 ProcessQueue，然后调用了 submitConsumeRequest 方法。跟并发消息处理方式不同的是，submitConsumeRequest 方法并没有处理拉取到的消息，而真正处理的时候是从 ProcessQueue 获取。

2.5 处理消息

处理消息的逻辑在 ConsumeMessageOrderlyService 的内部类 ConsumeRequest，这是一个线程类，run 方法如下：

public void run() {
 //省略部分逻辑
 //1.获取到 MessageQueueLock 对应的锁
 final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
 synchronized (objLock) {
  if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
   || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
   final long beginTime = System.currentTimeMillis();
   for (boolean continueConsume = true; continueConsume; ) {
    //省略延后执行的逻辑
    final int consumeBatchSize =
     ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
                //2.从 processQueue 拉取消息
    List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
    if (!msgs.isEmpty()) {
     final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);

     ConsumeOrderlyStatus status = null;
                    //省略部分逻辑
     boolean hasException = false;
     try {
         //3.获取处理锁
      this.processQueue.getConsumeLock().lock();
      //4.执行消费处理逻辑
      status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
     } catch (Throwable e) {
      log.warn(String.format("consumeMessage exception: %s Group: %s Msgs: %s MQ: %s",
       RemotingHelper.exceptionSimpleDesc(e),
       ConsumeMessageOrderlyService.this.consumerGroup,
       msgs,
       messageQueue), e);
      hasException = true;
     } finally {
         //5.释放处理锁
      this.processQueue.getConsumeLock().unlock();
     }
     //省略部分逻辑
     continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
    } else {
     continueConsume = false;
    }
   }
  } else {
   //省略部分逻辑
   ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
  }
 }
}

上面的代码总结一下，Consumer 消费消息的逻辑如下：

对 MessageQueueLock 进行加锁，这样就保证只有一个线程在处理当前 MessageQueue；
从 ProcessQueue 拉取一批消息；
获取 ProcessQueue 锁，这样保证了只有当前线程可以进行消息处理，同时也可以防止 Rebalance 线程把当前处理的 MessageQueue 移除掉；
执行消费处理逻辑；
释放 ProcessQueue 处理锁；6.processConsumeResult 方法更新消息偏移量。

注意：ProcessQueue 中的锁是 ReentrantLock。

3 重试

跟并发消息不一样的是，顺序消息消费失败后并不会把消息发送到 Broker，而是直接在 Consumer 端进行重试，如果重试次数超过了最大重试次数（16 次），则发送到 Broker，Broker 则将消息推入死信队列。如下图：

4 总结

RocketMQ 顺序消息的原理是在 Producer 端把一批需要保证顺序的消息发送到同一个 MessageQueue，Consumer 端则通过加锁的机制来保证消息消费的顺序性，Broker 端通过对 MessageQueue 进行加锁，保证同一个 MessageQueue 只能被同一个 Consumer 进行消费。

根据实现原理可以看到，RocketMQ 的顺序消息可能存在两个坑：

有顺序性的消息需要发送到同一个 MessageQueue，可能导致单个 MessageQueue 消息量很大，而 Consumer 端消费的时候只能单线程消费，很可能导致当前 MessageQueue 消息积压；
如果顺序消息 MessageQueue 所在的 broker 挂了，这时 Producer 只能把消息发送到其他 Broker 的 MessageQueue 上，而如果新的 MessageQueue 被其他 Consumer 消费，这样两个 Consumer 消费的消息就不能保证顺序性了。如下图：

Broker1 发生故障，把订单出库的消息发送到了 Broker2，由 Consumer2 来进行消费，消息顺序很可能会错乱。