消息生产者#xff0c;一般主题会由多个生产者组成#xff0c;生产者组。

消息消费者#xff0c;一般一个主题…1、消息消费需要解决的问题

首先再次重复啰嗦一下

1、一个消费组中多个消费者是如何对消息队列1个主题多个消息队列进行负载消费的。

2、一个消费者中多个线程又是如何协作并发的消费分配给该消费者的消息队列中的消息呢

3、消息消费进度如何保存包括MQ是如何知道消息是否正常被消费了。

4、RocketMQ

1个消费者可以消费多个消息队列但一个消息队列同一时间只能被一个消费者消费这又是如何实现的呢

本文紧接着上文消息消费概述

PullResult那问题来了pullRequestQueue

那我们就先沿着这条线索分析下去看一下

run()MQClientInstance.doRebalance()DefaultMQPulConsumerImpl.doRebalance()RebalanceImpl.doRebalance()RebalanceImpl.rebalanceByTopicRebalanceImpl.updateProcessQueueTableInRebalanceRebalanceImpl.dispatchPullRequestDefaultMQPushConsumerImpl.executePullRequestImmediately从上面可以直观的看出向

PullMesssageService

那么RebalanceService是何许人也让我们一起来揭开其神秘面纱。

2、消息消费负载机制分析

DefaultMQPushConsumerImpl.doRebalance

RebalanceImpl

到这里经过层层对象委托终于进入到实现消息负载分发的核心地带了RebalanceImpl

类我们应该停下脚步先重点认识一下RebalanceImpl类。

3、RebalanceImpl

消费模式。

AllocateMessageQueueStrategy

mqClientFactory

part1:根据消息消费模式集群还是广播我们先重点看集群模式。

part2:

获取主题的消息消费队列、主题与该消费组的消费者id列表,任意一个为空则退出方法的执行。

part3:

主要是对主题的消息队列排序、消费者ID进行排序然后利用分配算法计算当前消费者ID(mqClient.clientId)

具体的消息消费队列分配算法参考AllocateMessageQueueStrategy的实现类具体算法实现就不细化研究了。

有8个消息队列(q1,q2,q3,q4,q5,q6,q7,q8)

比如有三个消费者

不在当期主题的处理范围内由于消息队列数量变化等原因消费者的消费队列发生了变化该消息队列已经分配给别的消费者去消费了首先设置该消息队列为丢弃

(dropped

中拉取数据然后执行removeUnecessaryMessageQueue(mq,pq)

既然我们都是从Push进入的本文以Push模式展开(同时我们也可以先思考思考push,pull差别)移步到RebalancePushImpl。

MessageQueue

持久化然后丢弃重新被其他消费者加载。

顺序消息将会本系列的后续文章中详细介绍。

接下来处理

然后计算下一个拉取偏移量然后每一个MessageQueue创建一个拉取任务(PullRequest)。

RebalancePushImpl

pullRequestQueue中放入PullRequest,则PullMessageService线程

part5:如果消息负载发生变化需处理

这里主要看出来当消费者挂断后或主题消息队列动态变化后消息负载会发生变化的重新分布情况。

本文主要阐述了消息消费端负载机制这里消息非顺序消息机制就梳理到这里了大概再总结一下

1、首先RebalanceService线程启动为消费者分配消息队列其实每一个MessageQueue

会构建一个

pullMessage,通过网络从broker端拉取消息一次最多拉取的消息条数可配置默认为32条然后然后将拉取的消息执行过滤等然后封装成任务ConsumeRequest,提交到消费者的线程池去执行每次消费消息后又将该

PullRequest

PullMessageService中DefaultMQPushConsumerImpl

下文预告

CommitLog写入与ConsumeQueue队列的持久化机制

RocketMQ顺序消息