网站建设企业合作邀请函,google优化师,反向代理服务器做wordpress外网,2345网址导航电脑版1、消息消费需要解决的问题
首先再次重复啰嗦一下 RocketMQ 消息消费的一些基本元素的关系
主题 —》 消息队列(MessageQueue) 1 对多。
主题 —》 消息生产者#xff0c;一般主题会由多个生产者组成#xff0c;生产者组。
主题 —》 消息消费者#xff0c;一般一个主题…1、消息消费需要解决的问题
首先再次重复啰嗦一下 RocketMQ 消息消费的一些基本元素的关系
主题 —》 消息队列(MessageQueue) 1 对多。
主题 —》 消息生产者一般主题会由多个生产者组成生产者组。
主题 —》 消息消费者一般一个主题也会被多个消费者消费。
那消息消费至少需要解决如下问题
1、一个消费组中多个消费者是如何对消息队列1个主题多个消息队列进行负载消费的。
2、一个消费者中多个线程又是如何协作并发的消费分配给该消费者的消息队列中的消息呢
3、消息消费进度如何保存包括MQ是如何知道消息是否正常被消费了。
4、RocketMQ 推拉模式实现机制。
再提一个业界关于消费者与消息队列的消费规则。
1个消费者可以消费多个消息队列但一个消息队列同一时间只能被一个消费者消费这又是如何实现的呢
本文紧接着上文消息消费概述 。
继续探讨消息分发与消费端负载均衡。
我们从上文知道PullMessageService 线程主要是负责 pullRequestQueue 中的 PullResult那问题来了pullRequestQueue 中的数据从哪来在什么时候由谁来填充呢。 那我们就先沿着这条线索分析下去看一下 PullMessageService 的 pullReqestQueue 添加元素的方法的调用链条如下 也就是调用链
RebalanceService. run()MQClientInstance.doRebalance()DefaultMQPulConsumerImpl.doRebalance()RebalanceImpl.doRebalance()RebalanceImpl.rebalanceByTopicRebalanceImpl.updateProcessQueueTableInRebalanceRebalanceImpl.dispatchPullRequestDefaultMQPushConsumerImpl.executePullRequestImmediately从上面可以直观的看出向 PullMesssageService 的 LinkedBlockingQueue pullRequestQueue 添加 PullRequest的是 RebalanceService.run 方法就是向 PullMessageService 中放入 PullRequest,才会驱动 PullMessageSerivce run方法的运行如果 pullRequestQueue 中没有元素PullMessageService 线程将被阻塞。
那么RebalanceService是何许人也让我们一起来揭开其神秘面纱。
2、消息消费负载机制分析
2.1 RebalanceService 线程 从上面可以看出MQClientInstance 持有一个 RebalanceService 线程并启动它。RebalanceService 线程的 run 方法比较简单就是直接调用 mqClientFactory.doRebalance。
下面重点分步骤来详细探究 MQClientInstance.doRebalance 方法的执行流程。
2.1.1 MQClientInstance.doRebalance
循环遍历每个消费组获取 MQConsumeInner 对象其实就是 DefaultMQPushConsumerImpl 或 DefaultMQPullConsumerImpl 对象并执行其 doRebalance 方法。 2.1.2 DefaultMQPushConsumerImpl.doRebalance RebalanceImpl doRebalance 到这里经过层层对象委托终于进入到实现消息负载分发的核心地带了RebalanceImpl 类我们应该停下脚步先重点认识一下RebalanceImpl类。
3、RebalanceImpl 类初探
我们先来看看其核心属性 ConcurrentMapMessageQueue, ProcessQueue processQueueTable 消息处理队列。ConcurrentMapString, Set topicSubscribeInfoTable topic 的队列信息。ConcurrentMapString, SubscriptionData subscriptionInner 订阅信息。String consumerGroup 消费组名称。MessageModel messageModel 消费模式。AllocateMessageQueueStrategy allocateMessageQueueStrategy 队列分配算法。MQClientInstance mqClientFactory MQ 客户端实例。
下面还是从doRebalance方法入手 1、根据 topic 来进行负载。
2、移除 MessageQueue如果 MesageQueue 的 topic 不在订阅的主题中接下来重点关注 rebalanceByTopic 方法。
RebalanceImpl rebalanceByTopic详解:
part1:根据消息消费模式集群还是广播我们先重点看集群模式。
part2: 获取主题的消息消费队列、主题与该消费组的消费者id列表,任意一个为空则退出方法的执行。 part3: 主要是对主题的消息队列排序、消费者ID进行排序然后利用分配算法计算当前消费者ID(mqClient.clientId) 分配出需要拉取的消息队列。
具体的消息消费队列分配算法参考AllocateMessageQueueStrategy的实现类具体算法实现就不细化研究了。
在这里举一个最简单的队列分配机制比如一个topic 有8个消息队列(q1,q2,q3,q4,q5,q6,q7,q8) 比如有三个消费者 c1,c2,c3
一种队列负载算法 q1,q4,q7 分给c1,,q2,q5,q8 c2,,q3,q5 给 c3。下文会专题研究一下负载算法。 part4: 更新主题的消息消费处理队列并返回消息队列负载是否改变。 遍历消息队列-处理队列缓存只处理 mq 的主题与该主题相关的 ProcessQueue, 如果 mq 不在当期主题的处理范围内由于消息队列数量变化等原因消费者的消费队列发生了变化该消息队列已经分配给别的消费者去消费了首先设置该消息队列为丢弃 (dropped 为 voliate 修饰)可以及时的阻止继续向 ProceeQueue 中拉取数据然后执行removeUnecessaryMessageQueue(mq,pq) 来判断是否需要移除。 既然我们都是从Push进入的本文以Push模式展开(同时我们也可以先思考思考push,pull差别)移步到RebalancePushImpl。 目前只看非顺序消息逻辑就比较简单了丢弃之前先将 MessageQueue 消息消费进度 持久化然后丢弃重新被其他消费者加载。顺序消息将会本系列的后续文章中详细介绍。
接下来处理 MessageQueue 的 ProcessQueue,也就是在 ProcessQueueTable 中没有 mq 的处理队列因为重新负载后可能会分配一些新的队列。 主要就是在内存中移除 MessageQueue 的 offerset, 然后计算下一个拉取偏移量然后每一个MessageQueue创建一个拉取任务(PullRequest)。 RebalancePushImpl PullMessageService 往PullServiceMessage中的 pullRequestQueue中放入PullRequest,则PullMessageService线程 的run方法就不会阻塞 part5:如果消息负载发生变化需处理 主要是调整主题小各个队列的拉取阔值。 这里主要看出来当消费者挂断后或主题消息队列动态变化后消息负载会发生变化的重新分布情况。
总结 本文主要阐述了消息消费端负载机制这里消息非顺序消息机制就梳理到这里了大概再总结一下
1、首先RebalanceService线程启动为消费者分配消息队列其实每一个MessageQueue 会构建一个 PullRequest 对象然后通过 RebalanceImpl 将 PullRequest放入到 PullMessageService 线程的 LinkedBlockingQueue, 进而唤醒 queue.take()方法然后执行 DefaultMQPushConsumerImpl 的 pullMessage,通过网络从broker端拉取消息一次最多拉取的消息条数可配置默认为32条然后然后将拉取的消息执行过滤等然后封装成任务ConsumeRequest,提交到消费者的线程池去执行每次消费消息后又将该 PullRequest 放入到 PullMessageService中DefaultMQPushConsumerImpl 的机制就是pullInterval 为 0
下文预告
CommitLog写入与ConsumeQueue队列的持久化机制
消息消费进度存储机制再谈RocketMQ消息存储
RocketMQ顺序消息
RocketMQ主从机制
