RocketMQ

问题

背景是这样： 最近有个项目用MQ做消息流转，在RocketMQ集群模式下，一个消费者实例，订阅了两个Topic A、B。

Topic A：存储的是批量业务消息。
Topic B：存储的是单个业务消息。

有个小伙伴问我：A 先存入了100万消息，消费端线程开始消费 Topic A 数据，消费速度较慢；之后 B 也存入了1万消息，A 的数据积压会阻塞 B 的消费吗？

先说结论：B 消息消费会有一定的延迟，但不会绝对阻塞。

分析

本文 RocketMQ 版本：4.4.0。

从一张类图入手。

项目中，生产者 DefaultMQProducer 和消费者 DefaultMQPushConsumer 启动的时候，都会调用 MQClientInstance#start 方法来启动 MQClientInstance 组件。

MQClientInstance#start 主要工作：

启动NettyRemotingClient客户端，用于与NameServer和Broker进行通信。
定时更新Topic所对应路由信息、清除离线broker、向所有在线broker发送心跳包等。
启动拉取消息服务线程（PullMessageService），异步拉取消息。
启动负载均衡服务线程（RebalanceService），用于实现消息队列（Message Queue）的负载均衡。

在RocketMQ中，有ProcessQueue和MessageQueue两个概念。

MessageQueue表示一个消息队列，它包含主题、队列ID、Broker地址信息。

ProcessQueue 表示消费者正在消费的消息队列，是一个消费进度的抽象概念。在消费者消费消息时，将消息从MessageQueue中取出，并将消息保存在ProcessQueue中进行消费。包含了消费者消费的消息列表、消息偏移量以及一些状态信息等。

消费端 ProcessQueue 与 Broker 中的 MessageQueue 是一一对应的。

PullMessageService#run

在 PullMessageService 类中定义了 pullRequestQueue 属性，用于存储拉取消息请求（PullRequest）的队列。

PullRequest 拉取消息请求的数据结构定义了：

消费者组。
消费主题。
消费队列。
消费偏移量等。

PullMessageService#run 方法会从队列中取出 PullRequest 并执行，通过不断的发送 PullRequest来持续的拉取消息。

public class PullMessageService extends ServiceThread {
    private final LinkedBlockingQueue<PullRequest> pullRequestQueue = new LinkedBlockingQueue<PullRequest>();
  
  @Override
    public void run() {
      PullRequest pullRequest = this.pullRequestQueue.take();
                this.pullMessage(pullRequest);
    }
}

PullMessageService#pullMessage 方法找到消费组里面的消费实现类，执行对应的逻辑。

private void pullMessage(final PullRequest pullRequest) {
        final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
        if (consumer != null) {
            DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
            impl.pullMessage(pullRequest);
        } else {
            log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
        }
    }

DefaultMQPushConsumerImpl#pullMessage主要功能：

消费者服务状态校验，判断消费者是否是运行态；
流控校验，判断ProcessMessage 队列缓存消息数是否超过阀值。
并发消费和顺序消费相关的校验，并发消费限制最大跨度偏移量（offset），顺序消费则判断是否锁定，未锁定设置消费点位。
创建匿名内部类PullCallback，后续拉取消息返回响应后会回调onSuccess完成消息的消费；
调用PullAPIWrapper#pullKernelImpl拉取消息。

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
  public void pullMessage(final PullRequest pullRequest) {
    //校验
    //判断
    ...
    
    //创建匿名内部类
    PullCallback pullCallback = new PullCallback() {
      @Override
      public void onSuccess(PullResult pullResult) {
        switch (pullResult.getPullStatus()) {
          case FOUND:
            有新消息，封装请求提交到线程池，
            自定义消费者消费数据
            
            DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
              pullResult.getMsgFoundList(),
              processQueue,
              pullRequest.getMessageQueue(),
              dispatchToConsume);
          case NO_NEW_MSG:
          case NO_MATCHED_MSG:
            没有新消息或未匹配到消息时，继续往队列放入拉取消息请求（PullRequest），循环调用。
          ...
      }
    }
  }
}

PullStatus状态定义值，消息的中间状态，枚举：

FOUND：拉取到消息。
NO_NEW_MSG：没有新消息。
NO_MATCHED_MSG：没有匹配的消息。
OFFSET_ILLEGAL：非法的偏移量，可能设置的太大或大小。

ConsumeMessageConcurrentlyService#submitConsumeRequest 消费线程入口是这里，封装一个consumerRequest对象来执行业务初始化消费者线程池，会根据配置来创建消费线程数。

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
  //线程池，用于执行自定义消费逻辑
  private final ThreadPoolExecutor consumeExecutor;

  @Override
  public void submitConsumeRequest(
    提交任务，如定义了消费消息最大值时，会根据设置的值进行分割，然后提交到线程池待执行。
  )
  
  class ConsumeRequest implements Runnable {
    @Override
    public void run() {
      //找到自定义消息监听类
      MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
      ...
      //执行对应的实现类，消费数据
      status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
      ...
    }
  }
}

回到PullAPIWrapper#pullKernelImpl拉取消息方法中，最终会往Broker发送请求。

NettyRemotingAbstract#invokeAsyncImpl

public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
    final InvokeCallback invokeCallback){
    //获取信号量，最大为65535个
    boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
    if (acquired) {
      final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis - costTime, invokeCallback, once);
      //缓存正在执行的请求
      this.responseTable.put(opaque, responseFuture);
      
      channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture f) throws Exception {
            if (f.isSuccess()) {
                //设置响应状态
                responseFuture.setSendRequestOK(true);
                return;
            }
            requestFail(opaque);
            log.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
        }
      });
    }
}

上面代码中创建了ResponseFuture对象，放在了responseTable Map对象中，后续会获取后回调对应的方法。

也就是回调： MQClientAPIImpl#pullMessageAsync

private void pullMessageAsync(
    final String addr,
    final RemotingCommand request,
    final long timeoutMillis,
    final PullCallback pullCallback) {
  this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() {
    @Override
    public void operationComplete(ResponseFuture responseFuture) {
        RemotingCommand response = responseFuture.getResponseCommand();
        if (response != null) {
          PullResult pullResult = MQClientAPIImpl.this.processPullResponse(response);
          assert pullResult != null;
          //最终会回调PullMessageService#pullMessage方法中定义的回调函数，完成消息的分发
          pullCallback.onSuccess(pullResult);
        }
    }
  });
}

拉取到的消息会调用PullMessageService#pullMessage方法中定义的回调函数，完成消息的分发。

RebalanceService#run

定时任务，重分配消息队列。

public class RebalanceService extends ServiceThread {
@Override
    public void run() {
        while (!this.isStopped()) {
            //20s轮询
            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }
    }
}

最终会调用RebalanceImpl#updateProcessQueueTableInRebalance 重新分配 topic 队列。

updateProcessQueueTableInRebalance方法主要功能：

获取当前消费者的订阅信息，包括订阅的主题、消费者组等。
获取当前消费者的消费进度，包括已消费的消息队列和消息偏移量等。
获取当前主题的所有消息队列。
根据消费者组和消息队列数量进行负载均衡，将消息队列分配给消费者。
更新消费者的消费进度，将已消费的消息队列和消息偏移量更新为新分配的消息队列。
往pullRequestQueue队列中写入PullRequest请求数据。

最终也会生成拉取消息请求，放入队列中，等待PullMessageService线程执行。

答案？

至此，我们简单分析了消息的拉取、消费流程，那么回到我们之前的问题来。

假设单消费者实例定义了4个消费线程，那么线程池会创建4个核心线程用来执行任务。

生产者往Topic A先写入100万消息后，拉取线程会从这8个队列（Topic A和Topic B各4个）拉取消息，此时拉取到A的数据会放入消费者线程池等待消费者线程消费。

pull线程会持续的拉取，所以会持续的往线程池队列尾部写入Msg任务，按照我们分析的这种场景，Topic B消息虽然后面写入了，但是Topic B消息拉取后的数据放在了队列的中部或者后尾部的位置。

假如Thread 拿到消息后处理较慢，则会导致线程池队列的数据出现积压，也就是会最终会对B数据的消费产生影响，但不是绝对阻塞。

结论：

1. 会存在一定的消费延迟，但不是绝对的阻塞。也不是必须等A消费完，才会消费B的数据。

上测试代码

Topic A的消费类，线程睡眠1s，模拟正常业务处理耗时。

@Slf4j
@MQConsumeService(topic = "A")
public class A extends AbstractMQMsgProcessor {
    @Override
    protected boolean consumeMessage(String tag, String keys, MessageExt messageExt) {
        try {
            Thread.sleep(1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
        String message = new String(messageExt.getBody(), StandardCharsets.UTF_8);
        log.info("A message：{}, {}, {}, {}", message, Thread.currentThread().getName(), LocalDateTime.now(), keys);
        return true;
    }
}

Topic B的消费类。

@Slf4j
@MQConsumeService(topic = "B")
public class B extends AbstractMQMsgProcessor {
    @Override
    protected boolean consumeMessage(String tag, String keys, MessageExt messageExt) {
        String message = new String(messageExt.getBody(), StandardCharsets.UTF_8);
        log.info("B message：{}, {}, {}, {}", message, Thread.currentThread().getName(), LocalDateTime.now(), keys);
        return true;
    }
}

模拟先写入A类消息5000条，再写入B类消息100条。

@GetMapping(value = "/send")
public void reportList() throws Exception {
    for (int i = 0; i<5000; i++) {
        processor.aysncSend("A", "*", "你好" + i);
    }
    Thread.sleep(10000);
    for (int i = 0; i<100; i++) {
        processor.aysncSend("B", "*", "hello" + i);

    }
}