• RocketMQ消息消费（Consumer）源码解析

    RocketMQ中消息消费以消费组的模式开展，一个消费组内可以包含多个消费者，每一个消费者组可订阅多个主题，消费组内消费者之间有集群模式和广播模式两种消费模式。

    集群模式，主题下的同一条消息只允许被其中一个消费者消费。

    广播模式，主题下的同一条消息，将被集群内的所有消费者消费一次。

    消息服务器与消费者之间的消息传递也有两种模式：推模式、拉模式。

    所谓的拉模式即PullConsumer，是消费端主动拉起拉消息请求，
    而推模式即PushConsumer,是消息达到消息服务器后，推送给消息消费者。

    下面以PushConsumer模式为例，说明RocketMQ中消息消费的具体过程。

    Consumer 例子

    1. public class Consumer {
    2.  
    3.     public static void main(String[] args) throws InterruptedException, MQClientException {
    4.  
    5.         // Instantiate with specified consumer group name.
    6.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
    7.          
    8.         // Specify name server addresses.
    9.         consumer.setNamesrvAddr("localhost:9876");
    10.         
    11.         // Subscribe one more more topics to consume.
    12.         consumer.subscribe("TopicTest", "*");
    13.         // Register callback to execute on arrival of messages fetched from brokers.
    14.         consumer.registerMessageListener(new MessageListenerConcurrently() {
    15.  
    16.             @Override
    17.             public ConsumeConcurrentlyStatus consumeMessage(List msgs,
    18.                 ConsumeConcurrentlyContext context) {
    19.                 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
    20.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    21.             }
    22.         });
    23.  
    24.         //Launch the consumer instance.
    25.         consumer.start();
    26.  
    27.         System.out.printf("Consumer Started.%n");
    28.     }
    29. }

    首先，初始化一个DefaultMQPushConsumer，并设置group name。然后设置订阅的topic，一个consumer可以订阅多个topic。
    设置Listener，当新消息到来时会回调consumeMessage()方法，用户通过实现这个方法来做业务逻辑处理。
    最后启动consumer，开始接收消息。
    下面我们看下Consumer的启动过程的代码。

    Consumer启动

    DefaultPushConsumer是DefaultPushConsumerImpl的Facade类，启动直接调用DefaultPushConsumerImpl.start()方法

    1. public void start() throws MQClientException {
    2.         switch (this.serviceState) {
    3.             case CREATE_JUST:
    4.                 log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
    5.                     this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
    6.                 this.serviceState = ServiceState.START_FAILED;
    7.  
    8.                 // 检查配置
    9.                 this.checkConfig();
    10.  
    11.                 // Rebalance负载均衡 复制订阅数据
    12.                 this.copySubscription();
    13.  
    14.                 // 设置instanceName,为一个字符串化的数字,比如10072
    15.                 if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
    16.                     this.defaultMQPushConsumer.changeInstanceNameToPID();
    17.                 }
    18.  
    19.                 // 获取MQClient对象，clientId为ip@instanceName，比如192.168.0.1@10072
    20.                 this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
    21.  
    22.                 // 设置负载均衡器
    23.                 this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
    24.                 //默认这是消费模式为集群模式,每条消息被同一组的消费者中的一个消费
    25.                 //还可以设置为广播模式,每条消息被同一个组的所有消费者都消费一次
    26.                 this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
    27.                 //默认是AllocateMessageQueueAveragely,均分策略
    28.                 this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
    29.                 this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
    30.  
    31.                 // 拉取API封装
    32.                 this.pullAPIWrapper = new PullAPIWrapper(mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
    33.                 this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
    34.  
    35.                 //生成消费进度处理器,集群模式下消费进度保存在Broker上，因为同一组内的消费者要共享进度;广播模式下进度保存在消费者端
    36.                 if (this.defaultMQPushConsumer.getOffsetStore() != null) {
    37.                     this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
    38.                 } else {
    39.                     switch (this.defaultMQPushConsumer.getMessageModel()) {
    40.                         case BROADCASTING:
    41.                             this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
    42.                             break;
    43.                         case CLUSTERING:
    44.                             this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
    45.                             break;
    46.                         default:
    47.                             break;
    48.                     }
    49.                 }
    50.                 this.offsetStore.load();   //若是广播模式,加载本地的消费进度文件
    51.  
    52.                 // 根据监听是顺序模式还是并发模式来生成相应的ConsumerService
    53.                 if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
    54.                     this.consumeOrderly = true;
    55.                     this.consumeMessageService = new ConsumeMessageOrderlyService(this, (MessageListenerOrderly)this.getMessageListenerInner());
    56.                 } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
    57.                     this.consumeOrderly = false;
    58.                     this.consumeMessageService = new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently)this.getMessageListenerInner());
    59.                 }
    60.                 this.consumeMessageService.start();
    61.  
    62.                 // 设置MQClient对象
    63.                 boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
    64.                 if (!registerOK) {
    65.                     this.serviceState = ServiceState.CREATE_JUST;
    66.                     this.consumeMessageService.shutdown();
    67.                     throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
    68.                         + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
    69.                         null);
    70.                 }
    71.                 mQClientFactory.start();
    72.                 log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
    73.  
    74.                 // 设置服务状态
    75.                 this.serviceState = ServiceState.RUNNING;
    76.                 break;
    77.             case RUNNING:
    78.             case START_FAILED:
    79.             case SHUTDOWN_ALREADY:
    80.                 throw new MQClientException("The PushConsumer service state not OK, maybe started once, "//
    81.                     + this.serviceState//
    82.                     + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
    83.                     null);
    84.             default:
    85.                 break;
    86.         }
    87.  
    88.         //  从Namesrv获取TopicRouteData,更新TopicPublishInfo和MessageQueue   （在Consumer start时马上调用，之后每隔一段时间调用一次）
    89.         this.updateTopicSubscribeInfoWhenSubscriptionChanged();
    90.  
    91.         // 向TopicRouteData里的所有Broker发送心跳,注册Consumer/Producer信息到Broker上   （在Consumer start时马上调用，之后每隔一段时间调用一次）
    92.         this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    93.  
    94.         // 唤醒MessageQueue均衡服务,负载均衡后马上开启第一次拉取消息
    95.         this.mQClientFactory.rebalanceImmediately();
    96.     }

    1、函数copySubscription：将订阅信息跟RebalanceImpl同步，这个类是consumer的核心逻辑实现类。如果consumer是cluster模式，并且订阅了TopicA的消息，那客户端会自动订阅%RETRY%TopicA。那这个%RETRY%开头的topic是做什么的呢？我们知道consumer消费消息处理失败的话，broker是会延时一定的时间重新推送的，重新推送不是跟其它新消息一起过来，而是通过单独的%RETRY%的topic过来。

    2、getAndCreateMQClientInstance：初始化一个MQClientInstance，这个跟producer共用一个实现。

    3、rebalanceImpl：设置负载均衡器

    4、pullApiWrapper:封装实现类，封装了broker的通信接口

    MQClientInstance启动

    首先看下初始化的部分

    1.     public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
    2.         //前面的逻辑跟Producer相同
    3.         ...
    4.         ...
    5.         //1、Pull请求服务，异步发送请求到broker并负责将返回结果放到缓存队列
    6.         this.pullMessageService = new PullMessageService(this);
    7.         //2、定时或者被触发做subscribe queue的re-balance
    8.         this.rebalanceService = new RebalanceService(this);
    9.         //3、初始化一个自用的producer，`CLIENT_INNER_PRODUCER`
    10.         this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
    11.         this.defaultMQProducer.resetClientConfig(clientConfig);
    12.         ...
    13.     }

    其中第三步中的自用producer，主要用于在消费失败或者超时后发送重试的消息给broker。
    下面看下启动的过程中Consumer相关的部分：

    1. public void start() throws MQClientException {
    2.  
    3.         synchronized (this) {
    4.             switch (this.serviceState) {
    5.                 case CREATE_JUST:
    6.                     ...
    7.                     // 1、Start various schedule tasks
    8.                     this.startScheduledTask();
    9.                     // 2、Start pull service,开始处理PullRequest
    10.                     this.pullMessageService.start();
    11.                     // 3、Start rebalance service
    12.                     this.rebalanceService.start();
    13.                     // 4、Start push service，consumer预留的producer，发送要求重新的消息
    14.                     this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
    15.                     ...
    16.             }
    17.         }
    18.     }

    函数startScheduledTask启动的定时任务中，consumer相关的任务有两个

    1.         //保存消费进度，广播消息存在本地，集群消息上传到所有的broker
    2.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    3.  
    4.             @Override
    5.             public void run() {
    6.                 try {
    7.                     MQClientInstance.this.persistAllConsumerOffset();
    8.                 } catch (Exception e) {
    9.                     log.error("ScheduledTask persistAllConsumerOffset exception", e);
    10.                 }
    11.             }
    12.         }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
    13.         //对于`PushConsumer`，根据负载调整本地处理消息的线程池corePool大小
    14.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    15.  
    16.             @Override
    17.             public void run() {
    18.                 try {
    19.                     MQClientInstance.this.adjustThreadPool();
    20.                 } catch (Exception e) {
    21.                     log.error("ScheduledTask adjustThreadPool exception", e);
    22.                 }
    23.             }
    24.         }, 1, 1, TimeUnit.MINUTES);

    第3步，RebalanceService的任务主要是调用RebalanceImpl，来给consumer重新调整和分配queue。

    • 定时触发(20sec)做rebalance
    • 接口触发，1）收到broker的consumer list发生变化通知后需要重新做负载均衡，比如同一个group中新加入了consumer或者有consumer下线；2）consumer启动的时候

    从以上的PushConsumer启动逻辑可以看出，主要的消息读取逻辑都是由RebalanceImpl完成的，通过调用doRebalance()来触发，下面看下具体实现。

    RebalanceImpl触发Pull消息

    1. public void doRebalance(final boolean isOrder) {
    2.         //获取该consumer的订阅信息
    3.         Map subTable = this.getSubscriptionInner();
    4.         if (subTable != null) {
    5.             for (final Map.Entry entry : subTable.entrySet()) {
    6.                 final String topic = entry.getKey();
    7.                 try {
    8.                     //循环针对所有订阅的topic，做rebalance
    9.                     this.rebalanceByTopic(topic, isOrder);
    10.                 } catch (Throwable e) {
    11.                     if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
    12.                         log.warn("rebalanceByTopic Exception", e);
    13.                     }
    14.                 }
    15.             }
    16.         }
    17.         //做完rebalance后，检查是否有的queue已经不归自己负责消费，是的话就释放缓存message的queue
    18.         this.truncateMessageQueueNotMyTopic();
    19.     }

    主要的逻辑都是在rebalanceByTopic()中实现的：

    1. private void rebalanceByTopic(final String topic, final boolean isOrder) {
    2.         switch (messageModel) {
    3.             case BROADCASTING: {
    4.                 ...
    5.                 ...
    6.                 break;
    7.             }
    8.             case CLUSTERING: {
    9.                 //1、从路由信息中获取topic对应所有的Queue
    10.                 Set mqSet = this.topicSubscribeInfoTable.get(topic);
    11.                 //2、从broker获取所有同一个group的所有Consumer ID
    12.                 List cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
    13.                 if (null == mqSet) {
    14.                     if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
    15.                         log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
    16.                     }
    17.                 }
    18.  
    19.                 if (null == cidAll) {
    20.                     log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);
    21.                 }
    22.  
    23.                 if (mqSet != null && cidAll != null) {
    24.                     List mqAll = new ArrayList();
    25.                     mqAll.addAll(mqSet);
    26.                    //3、将MQ和cid都排好序
    27.                     Collections.sort(mqAll);
    28.                     Collections.sort(cidAll);
    29.  
    30.                     AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
    31.                     //4、按照初始化是指定的分配策略，获取分配的MQ列表
    32.                     List allocateResult = null;
    33.                     try {
    34.                         allocateResult = strategy.allocate(
    35.                             this.consumerGroup,
    36.                             this.mQClientFactory.getClientId(),
    37.                             mqAll,
    38.                             cidAll);
    39.                     } catch (Throwable e) {
    40.                         log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(),
    41.                             e);
    42.                         return;
    43.                     }
    44.  
    45.                     Set allocateResultSet = new HashSet();
    46.                     if (allocateResult != null) {
    47.                         allocateResultSet.addAll(allocateResult);
    48.                     }
    49.                     //5、更新rebalanceImpl中的processQueue用来缓存收到的消息，对于新加入的Queue，提交一次PullRequest
    50.                     boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
    51.                     if (changed) {
    52.                         ...
    53.                         ...
    54.                         //6、同步数据到broker，通过发送一次心跳实现
    55.                         this.messageQueueChanged(topic, mqSet, allocateResultSet);
    56.                     }
    57.                 }
    58.                 break;
    59.             }
    60.             default:
    61.                 break;
    62.         }
    63.     }

    第4步，同一个topic的消息会分布于集群内的多个broker的不同queue上。同一个group下面会有多个consumer，分配策略AllocateMessageQueueStrategy的作用就是计算当前consumer应该消费哪几个queue的消息。

    第5步，根据分配策略分配到queue之后，会查看是否是新增的queue，如果是则提交一次PullRequest去broker拉取消息。

    下面来看下分配策略和Pull请求的提交过程。

    Queue分配策略AllocateMessageQueueStrategy

    系统默认使用AVG策略（AllocateMessageQueueAveragely），就是将该topic所有Queue按照broker和queueId从小到大做排列，按照consumer的数量平均分成几份。然后每个consumer分到一份，按照consumer排序后的顺序来领取。代码实现如下：

    1.     @Override
    2.     public List allocate(String consumerGroup, String currentCID, List mqAll,
    3.         List cidAll) {
    4.         int index = cidAll.indexOf(currentCID);
    5.         int mod = mqAll.size() % cidAll.size();
    6.         //AVG size计算方法，mq数量<=consumer数量，size=1，这种情况是很少的
    7.         //否则size=mq数量/consumer数量，余数是几则前几个consumer的size+1,这样所有的queue都会有consumer消费
    8.         int averageSize =
    9.             mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
    10.                 + 1 : mqAll.size() / cidAll.size());
    11.         int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
    12.         int range = Math.min(averageSize, mqAll.size() - startIndex);
    13.         //从第一个consumer开始分配，每个分avgSize个连续的Queue，
    14.         for (int i = 0; i < range; i++) {
    15.             result.add(mqAll.get((startIndex + i) % mqAll.size()));
    16.         }
    17.         return result;
    18.     }

    RocketMQ提供其它的queue分配策略：

    • AVG_BY_CIRCLE， 跟AVG类似，只是分到的queue不是连续的。比如一共12个Queue，3个consumer，则第一个consumer接收queue1，4，7，9的消息。
    • CONSISTENT_HASH，使用一致性hash算法来分配Queue，用户需自定义虚拟节点的数量
    • MACHINE_ROOM，将queue先按照broker划分几个computer room，不同的consumer只消费某几个broker上的消息
    • CONFIG,用户启动时指定消费哪些Queue的消息

    提交Pull请求

    通过上面的策略分配到queue之后，RebalanceImpl通过updateProcessQueueTableInRebalance()方法来检查新加入queue并提交pull请求。

    1. private boolean updateProcessQueueTableInRebalance(final String topic, final Set mqSet,
    2.         final boolean isOrder) {
    3.         boolean changed = false;
    4.  
    5.         Iterator> it = this.processQueueTable.entrySet().iterator();
    6.         while (it.hasNext()) {
    7.             Entry next = it.next();
    8.             MessageQueue mq = next.getKey();
    9.             ProcessQueue pq = next.getValue();
    10.  
    11.             if (mq.getTopic().equals(topic)) {
    12.                 if (!mqSet.contains(mq)) {//不再消费这个Queue的消息
    13.                     pq.setDropped(true);
    14.                     if (this.removeUnnecessaryMessageQueue(mq, pq)) {//保存offset并
    15.                         it.remove();
    16.                         changed = true;
    17.                         log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
    18.                     }
    19.                 } else if (pq.isPullExpired()) {//超过max idle时间
    20.                     switch (this.consumeType()) {
    21.                         case CONSUME_ACTIVELY:
    22.                             break;
    23.                         case CONSUME_PASSIVELY:
    24.                             pq.setDropped(true);
    25.                             if (this.removeUnnecessaryMessageQueue(mq, pq)) {
    26.                                 it.remove();
    27.                                 changed = true;
    28.                                 log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",
    29.                                     consumerGroup, mq);
    30.                             }
    31.                             break;
    32.                         default:
    33.                             break;
    34.                     }
    35.                 }
    36.             }
    37.         }
    38.  
    39.         List pullRequestList = new ArrayList();
    40.         for (MessageQueue mq : mqSet) {
    41.             if (!this.processQueueTable.containsKey(mq)) {//如果是新加入的Queue
    42.                 if (isOrder && !this.lock(mq)) {
    43.                     log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
    44.                     continue;
    45.                 }
    46.                //从offset store中移除过时的数据
    47.                 this.removeDirtyOffset(mq);
    48.                 ProcessQueue pq = new ProcessQueue();
    49.                //获取起始消费offset
    50.                 long nextOffset = this.computePullFromWhere(mq);
    51.                 if (nextOffset >= 0) {
    52.                    //为新的Queue初始化一个ProcessQueue，用来缓存收到的消息
    53.                     ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
    54.                     if (pre != null) {
    55.                         log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
    56.                     } else {
    57.                         log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
    58.                         //对新加的queue初始化一个PullRequest
    59.                         PullRequest pullRequest = new PullRequest();
    60.                         pullRequest.setConsumerGroup(consumerGroup);
    61.                         pullRequest.setNextOffset(nextOffset);
    62.                         pullRequest.setMessageQueue(mq);
    63.                         pullRequest.setProcessQueue(pq);
    64.                         pullRequestList.add(pullRequest);
    65.                         changed = true;
    66.                     }
    67.                 } else {
    68.                     log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
    69.                 }
    70.             }
    71.         }
    72.         //分发pull request到PullMessageService,拉取消息
    73.         this.dispatchPullRequest(pullRequestList);
    74.         return changed;
    75.     }

    从以上的代码可以看出，RebalanceImpl每次都会检查分配到的queue列表，如果发现有新的queue加入，就会给这个queue初始化一个缓存队列，然后新发起一个PullRequest给PullMessageService执行。由此可见，新增的queue只有第一次Pull请求时RebalanceImpl发起的，后续请求是在broker返回数据后，处理线程发起的。

    消息拉取服务PullMessageService

    这个服务就是一个单独运行的线程，在收到Pull请求后异步执行。

    1.     private void pullMessage(final PullRequest pullRequest) {
    2.         final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
    3.         if (consumer != null) {
    4.             DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
    5.             impl.pullMessage(pullRequest);
    6.         } else {
    7.             log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
    8.         }
    9.     }

    上面的真正的实现是在DefaultMQPushConsumerImpl.pullMessage()里面。

    1. public void pullMessage(final PullRequest pullRequest) {
    2.         final ProcessQueue processQueue = pullRequest.getProcessQueue();
    3.         ...
    4.         ...
    5.         pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());
    6.         ...
    7.         ...
    8.         long cachedMessageCount = processQueue.getMsgCount().get();
    9.         long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024);
    10.         //1、如果堆积未处理的消息过多，则扔回PullMessageService,延时执行（默认50ms）
    11.         if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
    12.             this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
    13.             if ((queueFlowControlTimes++ % 1000) == 0) {
    14.                 log.warn(...);
    15.             }
    16.             return;
    17.         }
    18.         //2、如果堆积消息的size过大，同上面的逻辑
    19.         if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
    20.             this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
    21.             if ((queueFlowControlTimes++ % 1000) == 0) {
    22.                 log.warn(...);
    23.             }
    24.             return;
    25.         }
    26.         //3、无序消息，消息offset跨度过大，同上面的流控逻辑
    27.         if (!this.consumeOrderly) {
    28.             if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
    29.                 this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
    30.                 if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
    31.                     log.warn(...);
    32.                 }
    33.                 return;
    34.             }
    35.         } else {
    36.             if (processQueue.isLocked()) {
    37.                 if (!pullRequest.isLockedFirst()) {
    38.                     final long offset = this.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());
    39.                     boolean brokerBusy = offset < pullRequest.getNextOffset();
    40.                     log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",
    41.                         pullRequest, offset, brokerBusy);
    42.                     if (brokerBusy) {
    43.                         log.info("[NOTIFYME]the first time to pull message, but pull request offset larger than broker consume offset. pullRequest: {} NewOffset: {}",
    44.                             pullRequest, offset);
    45.                     }
    46.  
    47.                     pullRequest.setLockedFirst(true);
    48.                     pullRequest.setNextOffset(offset);
    49.                 }
    50.             } else {
    51.                 this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
    52.                 log.info("pull message later because not locked in broker, {}", pullRequest);
    53.                 return;
    54.             }
    55.         }
    56.         //4、检查订阅关系有没有变化，有可能在延时期间，topic或者consumer的配置都发生了变化
    57.         final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
    58.         if (null == subscriptionData) {
    59.             this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
    60.             log.warn("find the consumer's subscription failed, {}", pullRequest);
    61.             return;
    62.         }
    63.  
    64.         final long beginTimestamp = System.currentTimeMillis();
    65.         //5、Pull Command发送后，返回结果处理
    66.         PullCallback pullCallback = new PullCallback() {
    67.             @Override
    68.             public void onSuccess(PullResult pullResult) {
    69.                 if (pullResult != null) {
    70.                     //6、消息预处理，客户端再次过滤，set minOffset和maxOffset
    71.                     pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
    72.                         subscriptionData);
    73.  
    74.                     switch (pullResult.getPullStatus()) {
    75.                         case FOUND:
    76.                             long prevRequestOffset = pullRequest.getNextOffset();
    77.                             pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    78.                             long pullRT = System.currentTimeMillis() - beginTimestamp;
    79.                             DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
    80.                                 pullRequest.getMessageQueue().getTopic(), pullRT);
    81.  
    82.                             long firstMsgOffset = Long.MAX_VALUE;
    83.                             //7、如果获取到的消息数为0，则立即发起下一次pull
    84.                             if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
    85.                                 DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
    86.                             } else {
    87.                                 firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();
    88.  
    89.                                 DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
    90.                                     pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
    91.                                 //8、消息放入ProcessQueue
    92.                                 boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
    93.                                 //9、消费消息，调用messageListener处理，处理完成会通知ProcessQueue
    94.                                 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
    95.                                     pullResult.getMsgFoundList(),
    96.                                     processQueue,
    97.                                     pullRequest.getMessageQueue(),
    98.                                     dispatchToConsume);
    99.                                 //10、再次提交pull request
    100.                                 if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
    101.                                     DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
    102.                                         DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
    103.                                 } else {
    104.                                     DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
    105.                                 }
    106.                             }
    107.  
    108.                             if (pullResult.getNextBeginOffset() < prevRequestOffset
    109.                                 || firstMsgOffset < prevRequestOffset) {
    110.                                 log.warn(
    111.                                     "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
    112.                                     pullResult.getNextBeginOffset(),
    113.                                     firstMsgOffset,
    114.                                     prevRequestOffset);
    115.                             }
    116.  
    117.                             break;
    118.                         case NO_NEW_MSG:
    119.                             pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    120.  
    121.                             DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
    122.  
    123.                             DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
    124.                             break;
    125.                         case NO_MATCHED_MSG:
    126.                             pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    127.  
    128.                             DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
    129.  
    130.                             DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
    131.                             break;
    132.                         case OFFSET_ILLEGAL://Queue已经不存在了
    133.                             log.warn("the pull request offset illegal, {} {}",
    134.                                 pullRequest.toString(), pullResult.toString());
    135.                             pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    136.  
    137.                             pullRequest.getProcessQueue().setDropped(true);
    138.                             DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {
    139.                                 //存储消费offset，从rebalance中移除ProcessQueue
    140.                                 @Override
    141.                                 public void run() {
    142.                                     try {
    143.                                         DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(),
    144.                                             pullRequest.getNextOffset(), false);
    145.  
    146.                                         DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());
    147.  
    148.                                         DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());
    149.  
    150.                                         log.warn("fix the pull request offset, {}", pullRequest);
    151.                                     } catch (Throwable e) {
    152.                                         log.error("executeTaskLater Exception", e);
    153.                                     }
    154.                                 }
    155.                             }, 10000);
    156.                             break;
    157.                         default:
    158.                             break;
    159.                     }
    160.                 }
    161.             }
    162.  
    163.             @Override
    164.             public void onException(Throwable e) {
    165.                 if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
    166.                     log.warn("execute the pull request exception", e);
    167.                 }
    168.  
    169.                 DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
    170.             }
    171.         };
    172.  
    173.         boolean commitOffsetEnable = false;
    174.         long commitOffsetValue = 0L;
    175.         if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) {
    176.             commitOffsetValue = this.offsetStore.readOffset(pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY);
    177.             if (commitOffsetValue > 0) {
    178.                 commitOffsetEnable = true;
    179.             }
    180.         }
    181.  
    182.         String subExpression = null;
    183.         boolean classFilter = false;
    184.         SubscriptionData sd = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
    185.         if (sd != null) {
    186.             if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {
    187.                 subExpression = sd.getSubString();
    188.             }
    189.  
    190.             classFilter = sd.isClassFilterMode();
    191.         }
    192.  
    193.         int sysFlag = PullSysFlag.buildSysFlag(
    194.             commitOffsetEnable, // commitOffset
    195.             true, // suspend
    196.             subExpression != null, // subscription
    197.             classFilter // class filter
    198.         );
    199.         try {
    200.             this.pullAPIWrapper.pullKernelImpl(
    201.                 pullRequest.getMessageQueue(),
    202.                 subExpression,
    203.                 subscriptionData.getExpressionType(),
    204.                 subscriptionData.getSubVersion(),
    205.                 pullRequest.getNextOffset(),
    206.                 this.defaultMQPushConsumer.getPullBatchSize(),
    207.                 sysFlag,
    208.                 commitOffsetValue,
    209.                 BROKER_SUSPEND_MAX_TIME_MILLIS,
    210.                 CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
    211.                 CommunicationMode.ASYNC,
    212.                 pullCallback
    213.             );
    214.         } catch (Exception e) {//错误处理，延时重试(默认3sec)
    215.             log.error("pullKernelImpl exception", e);
    216.             this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
    217.         }
    218.     }

    以上逻辑中主要关注第5步，在消息返回后，会将消息放入ProcessQueue，然后通知ConsumeMessageService来异步处理消息，然后再次提交Pull请求。这样对于用户端来说，只有ConsumeMessageService回调listener这一步是可见的，其它都是透明的。

    消息处理ConsumeMessageService

    消息处理的逻辑比较简单，就是回调Consumer启动时注册的Listener。无论Listener是否处理成功，消息都会从ProcessQueue中移除掉。我们看下对于Listener返回结果的处理方法。
    ConsumeMessageConcurrentlyService.processConsumeResult()

    1. /**
    2.      * 处理消费结果
    3.      *
    4.      * @param status         消费结果
    5.      * @param context        消费Context
    6.      * @param consumeRequest 提交请求
    7.      */
    8.     public void processConsumeResult(
    9.         final ConsumeConcurrentlyStatus status,
    10.         final ConsumeConcurrentlyContext context,
    11.         final ConsumeRequest consumeRequest
    12.     ) {
    13.         int ackIndex = context.getAckIndex();
    14.  
    15.         if (consumeRequest.getMsgs().isEmpty())
    16.             return;
    17.  
    18.         ...
    19.         ...
    20.  
    21.         switch (this.defaultMQPushConsumer.getMessageModel()) {
    22.             //broadcast模式，处理失败，不做处理
    23.             case BROADCASTING:
    24.                 for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
    25.                     MessageExt msg = consumeRequest.getMsgs().get(i);
    26.                     log.warn("BROADCASTING, the message consume failed, drop it, {}", msg.toString());
    27.                 }
    28.                 break;
    29.             case CLUSTERING:
    30.                 List msgBackFailed = new ArrayList(consumeRequest.getMsgs().size());
    31.                 for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
    32.                     MessageExt msg = consumeRequest.getMsgs().get(i);
    33.                     //Cluster模式，将消息发回broker重新发送
    34.                     boolean result = this.sendMessageBack(msg, context);
    35.                     if (!result) {
    36.                         msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
    37.                         msgBackFailed.add(msg);
    38.                     }
    39.                 }
    40.  
    41.                 if (!msgBackFailed.isEmpty()) {
    42.                     consumeRequest.getMsgs().removeAll(msgBackFailed);
    43.                     //发回broker失败，则再次尝试本地消费
    44.                     this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());
    45.                 }
    46.                 break;
    47.             default:
    48.                 break;
    49.         }
    50.         //将消费前缓存的消息清除
    51.         long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
    52.         //更新offset
    53.         if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
    54.             this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
    55.         }
    56.     }

    消息处理失败后，consumer会将消息发给broker，broker会根据重试次数来重新投递消息。sendback方法的实现如下

    1.     public boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context) {
    2.         int delayLevel = context.getDelayLevelWhenNextConsume();
    3.  
    4.         try {
    5.             this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, context.getMessageQueue().getBrokerName());
    6.             return true;
    7.         } catch (Exception e) {
    8.             log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);
    9.         }
    10.  
    11.         return false;
    12.     }

    首先会根据这是第几次消费失败，设置延时多长时间重新投递，然后调用DefaultMQPushConsumerImpl.sendMessageBack()的方法。默认设置下，最多会重新投递16次。

    1.     //consumer把没有消费的消息提交给broker，broker会延时一段时间后重新发送
    2.     public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName)
    3.         throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
    4.         try {
    5.             String brokerAddr = (null != brokerName) ? this.mQClientFactory.findBrokerAddressInPublish(brokerName)
    6.                 : RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());
    7.             //首先尝试直接发送CONSUMER_SEND_MSG_BACK命令给broker
    8.             this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg,
    9.                 this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000, getMaxReconsumeTimes());
    10.         } catch (Exception e) {
    11.             log.error("sendMessageBack Exception, " + this.defaultMQPushConsumer.getConsumerGroup(), e);
    12.             //如果发送失败，则把消息发送到%RETRY%topic，重新发送
    13.             Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup()), msg.getBody());
    14.  
    15.             String originMsgId = MessageAccessor.getOriginMessageId(msg);
    16.             MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
    17.  
    18.             newMsg.setFlag(msg.getFlag());
    19.             MessageAccessor.setProperties(newMsg, msg.getProperties());
    20.             MessageAccessor.putProperty(newMsg, MessageConst.PROPERTY_RETRY_TOPIC, msg.getTopic());
    21.             MessageAccessor.setReconsumeTime(newMsg, String.valueOf(msg.getReconsumeTimes() + 1));
    22.             MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(getMaxReconsumeTimes()));
    23.             newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
    24.  
    25.             this.mQClientFactory.getDefaultMQProducer().send(newMsg);
    26.         }
    27.     }

    一共有两种方式让broker重发，先尝试给broker发送send_msg_back的命令，如果失败了，则通过consumer预留的producer给%RETRY%topic发送消息，前面consumer启动的时候已经讲过，所有consumer都订阅%RETRY%topic，所以等于是自己给自己发一条消息。

  • 相关阅读:
    短时间内防止多次点击
    【Python/Pytorch - 网络模型】-- 高阶SVD算法
    60页5G+智慧农业大数据 2022
    基于node.js+Vue学院会议纪要管理系统 element
    Python configparser模块详解：配置文件解析利器
    这份阿里强推的并发编程知识点笔记，将是你拿大厂offer的突破口
    【Head First 设计模式】-- 观察者模式
    高通平台Android 蓝牙调试和配置手册-- Bluetooth ON/OFF Failures
    CP&FT测试介绍
    Kratos战神微服务框架（二）
  • 原文地址：https://blog.csdn.net/bao2901203013/article/details/126474661