RocketMQ 消费

发表于 | 分类于

6. 消息拉取并消费

本章我们将分别从Consumer和Broker两个地方进行介绍。

Consumer 处理

在第四章初始化与启动中，介绍了Consumer的初始化启动过程，其中会启动一个RebalanceService服务。它的作用就是定时去做重负载。
重负载的内容就是去Broker去查询当前所有的消费者然后根据消息分配策略计算出分配给自己的队列有哪些。最后跟本地进行比较，如果是新增的就发起消息的消费，如果是本地不再存在的则从本地删除，其他的则不变。
对于Consumer刚启动的时候，也会触发一次重负载。
重负载的逻辑在RebalanceImpl.rebalanceByTopic方法中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
private void rebalanceByTopic(final String topic, final boolean isOrder) {
   switch (messageModel) {
   case CLUSTERING: {
       // 获取topic下的所有queue
       Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
       //获取本消费组下的所有消费者
       List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
       if (mqSet != null && cidAll != null) {
           List<MessageQueue> mqAll = new ArrayList<>();
           mqAll.addAll(mqSet);
           Collections.sort(mqAll);
           Collections.sort(cidAll);
           AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
           // 根据 队列分配策略 分配消费队列
           List<MessageQueue> allocateResult;
           // 根据指定分配策略分配出给本consumer的消费队列
           allocateResult = strategy.allocate(this.consumerGroup, this.mQClientFactory.getClientId(), mqAll, cidAll);
           Set<MessageQueue> allocateResultSet = new HashSet<>();
           if (allocateResult != null) {
               allocateResultSet.addAll(allocateResult);
           }
           // 跟据最新的分配结果去更新consumer本地的消费队列
           boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
       }
       break;
   }
       case //...
   }
}

private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet, final boolean isOrder) {
   boolean changed = false;
   Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
   while (it.hasNext()) {
       // 筛选出不在最新消费队列中的本地队列进行移除，另外对于拉取超时的也同样移除
       Entry<MessageQueue, ProcessQueue> next = it.next();
       MessageQueue mq = next.getKey();
       ProcessQueue pq = next.getValue();
       if (mq.getTopic().equals(topic)) {
           if (!mqSet.contains(mq)) {
               pq.setDropped(true);
               if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                   it.remove();
                   changed = true;
                   log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
               }
           } else if (pq.isPullExpired()) {
               switch (this.consumeType()) {
                   case CONSUME_ACTIVELY:
                       break;
                   case CONSUME_PASSIVELY:
                       pq.setDropped(true);
                       if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                           it.remove();
                           changed = true;
                           log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",
                               consumerGroup, mq);
                       }
                       break;
                   default:
                       break;
               }
           }
       }
   }
   // 筛选出新加的
   List<PullRequest> pullRequestList = new ArrayList<>();
   for (MessageQueue mq : mqSet) {
       if (!this.processQueueTable.containsKey(mq)) {
           // 有序消费的时候，需要锁定记录
           if (isOrder && !this.lock(mq)) {
               log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
               continue;
           }
           this.removeDirtyOffset(mq);
           ProcessQueue pq = new ProcessQueue();
           long nextOffset = this.computePullFromWhere(mq);
           if (nextOffset >= 0) {
               ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
               if (pre != null) {
                   log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
               } else {
                   log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
                   PullRequest pullRequest = new PullRequest();
                   pullRequest.setConsumerGroup(consumerGroup);
                   pullRequest.setNextOffset(nextOffset);
                   pullRequest.setMessageQueue(mq);
                   pullRequest.setProcessQueue(pq);
                   pullRequestList.add(pullRequest);
                   changed = true;
               }
           }
       }
   }
   // 发起消息拉取请求
   this.dispatchPullRequest(pullRequestList);
   return changed;
}

特别需要说的是：对于有序消费，在重负载的时候会尝试锁住资源；如果锁资源失败则不进行消费。
对于新增的消费队列，consumer会构造拉取参数PullRequest调用dispatchPullRequest进行提交消费。可以看到，最终是将拉取请求丢进PullMessageService的消息拉取队列中pullRequestQueue。

1
2
3
4
5
6
7
8
9
10
11
public void dispatchPullRequest(List<PullRequest> pullRequestList) {
   for (PullRequest pullRequest : pullRequestList) {
       this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
   }
}
public void executePullRequestImmediately(final PullRequest pullRequest) {
   this.mQClientFactory.getPullMessageService().executePullRequestImmediately(pullRequest);
}
public void executePullRequestImmediately(final PullRequest pullRequest) {
   this.pullRequestQueue.put(pullRequest);
}

PullMessageService中有个后台线程会从pullRequestQueue不停取PullRequest开始拉取消息。
拉取消费过程大致如下：

  1. 调用pullAPIWrapper.pullKernelImpl去拉取消息
  2. 对拉取回来的消息调用PullCallback进行消费

拉取消息请求的是Broker的PullMessageProcessor，后面我们再来分析它。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
if (pullResult != null) {
     pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
         subscriptionData);
     switch (pullResult.getPullStatus()) {
         case FOUND:
             // 设置下一次消费位移
             pullRequest.setNextOffset(pullResult.getNextBeginOffset());
             // ... 统计
             if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
                 DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
             } else {
                 // 将无需的消息放到ProcessQueue中按offset排序
                 boolean dispathToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
                 // 提交消费请求，对于顺序有要求的请求提交到 ConsumeMessageOrderlyService中，没有要求的提交到ConsumeMessageConcurrentlyService进行消费
                 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(//
                     pullResult.getMsgFoundList(), //
                     processQueue, //
                     pullRequest.getMessageQueue(), //
                     dispathToConsume);
                 // 将pullRequest再丢进PullMessageService的队列中继续拉取
                 if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
                     DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
                         DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
                 } else {
                     DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                 }
             }
             break;
         case //...
     }
 }

Broker 处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend)
        throws RemotingCommandException {
   RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
   final PullMessageResponseHeader responseHeader = (PullMessageResponseHeader) response.readCustomHeader();
   final PullMessageRequestHeader requestHeader =
       (PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
   response.setOpaque(request.getOpaque());

   //... 一顿校验
   final boolean hasSuspendFlag = PullSysFlag.hasSuspendFlag(requestHeader.getSysFlag()); // 是否挂起请求，当没有消息时
   final boolean hasCommitOffsetFlag = PullSysFlag.hasCommitOffsetFlag(requestHeader.getSysFlag()); // 是否提交消费进度
   final boolean hasSubscriptionFlag = PullSysFlag.hasSubscriptionFlag(requestHeader.getSysFlag()); // 是否过滤订阅表达式(subscription)
   final long suspendTimeoutMillisLong = hasSuspendFlag ? requestHeader.getSuspendTimeoutMillis() : 0; // 挂起请求超时时长
   //... 一顿校验
   // 拉取信息
   final GetMessageResult getMessageResult = this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
           requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), subscriptionData);
   if (getMessageResult != null) {
       response.setRemark(getMessageResult.getStatus().name());

       responseHeader.setNextBeginOffset(getMessageResult.getNextBeginOffset());
       responseHeader.setMinOffset(getMessageResult.getMinOffset());
       responseHeader.setMaxOffset(getMessageResult.getMaxOffset());

       // 计算建议读取brokerId
       if (getMessageResult.isSuggestPullingFromSlave()) {
           responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
       } else {
           responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
       }

       switch (this.brokerController.getMessageStoreConfig().getBrokerRole()) {
       case ASYNC_MASTER:
       case SYNC_MASTER:
           break;
       case SLAVE:
           if (!this.brokerController.getBrokerConfig().isSlaveReadEnable()) { // 从节点不允许读取，告诉consumer读取主节点。
               response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
               responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
           }
           break;
       }
       //当消费过慢的时候，让其从Slave拉取数据
       if (this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
           // consume too slow ,redirect to another machine
           if (getMessageResult.isSuggestPullingFromSlave()) {
               responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
           }
           // consume ok
           else {
               responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
           }
       }
   }
  // ...

拉取消息后，需要进行一次判断：

  1. 当拉取到数据后，如果配置不通过堆传输的话，就通过zero-copy的方式直接写socket返回
  2. 如果没有拉取到数据，则将请求挂起，放到PullRequestHoldService中。当有新消息产生时，再返回。长轮询
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
switch (response.getCode()) {
 case ResponseCode.SUCCESS:
     if (this.brokerController.getBrokerConfig().isTransferMsgByHeap()) { // 内存中
         final long beginTimeMills = this.brokerController.getMessageStore().now();
         final byte[] r = this.readGetMessageResult(getMessageResult, requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());
         this.brokerController.getBrokerStatsManager().incGroupGetLatency(requestHeader.getConsumerGroup(),
             requestHeader.getTopic(), requestHeader.getQueueId(),
             (int) (this.brokerController.getMessageStore().now() - beginTimeMills));
         response.setBody(r);
     } else {
         // 拉取到信息后，通过zero-copy的方式发送数据
         FileRegion fileRegion = new ManyMessageTransfer(response.encodeHeader(getMessageResult.getBufferTotalSize()), getMessageResult);
         channel.writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
             @Override
             public void operationComplete(ChannelFuture future) throws Exception {
                 getMessageResult.release();
                 if (!future.isSuccess()) {
                     LOG.error("Fail to transfer messages from page cache to {}", channel.remoteAddress(), future.cause());
                 }
             }
         });
         response = null;
     }
     break;
 case ResponseCode.PULL_NOT_FOUND:
     if (brokerAllowSuspend && hasSuspendFlag) {
         // 允许挂起时，将请求挂起在PullRequestHoldService中。待有新消息产生时返回
         long pollingTimeMills = suspendTimeoutMillisLong;
         if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
             pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
         }
         String topic = requestHeader.getTopic();
         long offset = requestHeader.getQueueOffset();
         int queueId = requestHeader.getQueueId();
         PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills,
             this.brokerController.getMessageStore().now(), offset, subscriptionData);
         this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
         response = null;
         break;
     }
 case //...
}

最后，保存Topic下该Queue的消费进度。

1
2
3
4
5
6
7
8
9
10
 // 保存 消费进度
boolean storeOffsetEnable = brokerAllowSuspend;
storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag;
storeOffsetEnable = storeOffsetEnable && this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE;
if (storeOffsetEnable) {
  this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),
      requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());
}
return response;
}
您的支持是我创作源源不断的动力