rocketmq消费负载均衡--push消费详解

前言

本文介绍了DefaultMQPushConsumerImpl消费者,客户端负载均衡相关知识点。本文从DefaultMQPushConsumerImpl启动过程到实现负载均衡,从源代码一步一步分析,共分为6个部分进行介绍,其中第6个部分 rebalanceByTopic 为负载均衡的核心逻辑模块,具体过程运用了图文进行阐述。

介绍之前首先抛出几个问题:

1. 要做负载均衡,首先要解决的一个问题是什么?

2. 负载均衡是Client端处理还是Broker端处理?

个人理解:

1. 要做负载均衡,首先要做的就是信号收集。

所谓信号收集,就是得知道每一个consumerGroup有哪些consumer,对应的topic是谁。信号收集分为Client端信号收集与Broker端信号收集两个部分。

2. 负载均衡放在Client端处理。

具体做法是:消费者客户端在启动时完善rebalanceImpl实例,同时拷贝订阅信息存放rebalanceImpl实例对象中,另外也是很重要的一个步骤 -- 通过心跳消息,不停的上报自己到所有Broker,注册RegisterConsumer,等待上述过程准备好之后在Client端不断执行的负载均衡服务线程从Broker端获取一份全局信息(该consumerGroup下所有的消费Client),然后分配这些全局信息,获取当前客户端分配到的消费队列。

本文具体的内容:

I. copySubscription

Client端信号收集,拷贝订阅信息。

在DefaultMQPushConsumerImpl.start()时,会将消费者的topic订阅关系设置到rebalanceImpl的SubscriptionInner的map中用于负载:

private void copySubscription() throws MQClientException {
try {
//注:一个consumer对象可以订阅多个topic
Map<String, String> sub = this.defaultMQPushConsumer.getSubscription();
if (sub != null) {
for (final Map.Entry<String, String> entry : sub.entrySet()) {
final String topic = entry.getKey();
final String subString = entry.getValue();
SubscriptionData subscriptionData =
FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),//
topic, subString);
this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
}
}
if (null == this.messageListenerInner) {
this.messageListenerInner = this.defaultMQPushConsumer.getMessageListener();
}
switch (this.defaultMQPushConsumer.getMessageModel()) {
case BROADCASTING:
break;
case CLUSTERING:
final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
SubscriptionData subscriptionData =
FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),//
retryTopic, SubscriptionData.SUB_ALL);
this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
break;
default:
break;
}
}
catch (Exception e) {
throw new MQClientException("subscription exception", e);
}
}

FilterAPI.buildSubscriptionData接口将订阅关系转换为SubscriptionData 数据,其中subString包含订阅tag等信息。另外,如果该消费者的消费模式为集群消费,则会将retry的topic一并放到。

II. 完善rebalanceImpl实例

Client继续收集信息:

this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer
.getAllocateMessageQueueStrategy());
this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);

本文以DefaultMQPushConsumerImpl为例,因此this对象类型为DefaultMQPushConsumerImp。

III. this.rebalanceService.start()

开启负载均衡服务。this.rebalanceService是一个RebalanceService实例对象,它继承与ServiceThread,是一个线程类。 this.rebalanceService.start()执行时,也即执行RebalanceService线程体:

@Override
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStoped()) {
this.waitForRunning(WaitInterval);
this.mqClientFactory.doRebalance();
}
log.info(this.getServiceName() + " service end");
}

IV. this.mqClientFactory.doRebalance

客户端遍历消费组table,对该客户端上所有消费者独立进行负载均衡,分发消费队列:

public void doRebalance() {
for (String group : this.consumerTable.keySet()) {
MQConsumerInner impl = this.consumerTable.get(group);
if (impl != null) {
try {
impl.doRebalance();
} catch (Exception e) {
log.error("doRebalance exception", e);
}
}
}
}

V. MQConsumerInner.doRebalance

由于本文以DefaultMQPushConsumerImpl消费过程为例,即DefaultMQPushConsumerImpl.doRebalance:

@Override
public void doRebalance() {
if (this.rebalanceImpl != null) {
this.rebalanceImpl.doRebalance();
}
}

步骤II 中完善了rebalanceImpl实例,为调用rebalanceImpl.doRebalance()提供了初始数据。

rebalanceImpl.doRebalance()过程如下:

public void doRebalance() {
     // 前文copySubscription中初始化了SubscriptionInner
Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
if (subTable != null) {
for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
final String topic = entry.getKey();
try {
this.rebalanceByTopic(topic);
} catch (Exception e) {
if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
log.warn("rebalanceByTopic Exception", e);
}
}
}
}
this.truncateMessageQueueNotMyTopic();
}

VI. rebalanceByTopic -- 核心步骤之一

rebalanceByTopic方法中根据消费者的消费类型为BROADCASTING或CLUSTERING做不同的逻辑处理。CLUSTERING逻辑包括BROADCASTING逻辑,本部分只介绍集群消费负载均衡的逻辑。

集群消费负载均衡逻辑主要代码如下(省略了log等代码):

//1.从topicSubscribeInfoTable列表中获取与该topic相关的所有消息队列
Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
//2. 从broker端获取消费该消费组的所有客户端clientId
List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
f (null == mqSet) { ... }
if (null == cidAll) { ... }
if (mqSet != null && cidAll != null) {
List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
mqAll.addAll(mqSet);
Collections.sort(mqAll);
Collections.sort(cidAll);

     // 3.创建DefaultMQPushConsumer对象时默认设置为AllocateMessageQueueAveragely
AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

List<MessageQueue> allocateResult = null;
try {
         // 4.调用AllocateMessageQueueAveragely.allocate方法,获取当前client分配消费队列
allocateResult = strategy.allocate(
this.consumerGroup,
this.mQClientFactory.getClientId(),
mqAll,
cidAll);
} catch (Throwable e) {
return;
}
    // 5. 将分配得到的allocateResult 中的队列放入allocateResultSet 集合
Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
if (allocateResult != null) {
allocateResultSet.addAll(allocateResult);
}
、
     //6. 更新updateProcessQueue
boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet);
if (changed) {
this.messageQueueChanged(topic, mqSet, allocateResultSet);
}
}

注:BROADCASTING逻辑只包含上述的1、6。

集群消费负载均衡逻辑中的1、2、4这三个点相关知识为其核心过程,各个点相关知识如下:

第1点:从topicSubscribeInfoTable列表中获取与该topic相关的所有消息队列

第2点: 从broker端获取消费该消费组的所有客户端clientId

首先,消费者对象不断地向所有broker发送心跳包,上报自己,注册并更新订阅关系以及客户端ChannelInfoTable;之后,客户端在做消费负载均衡时获取那些消费客户端,对这些客户端进行负载均衡,分发消费的队列。具体过程如下图所示:

第4点:调用AllocateMessageQueueAveragely.allocate方法,获取当前client分配消费队列

注:上图中cId1、cId2、...、cIdN通过 getConsumerIdListByGroup 获取,它们在这个ConsumerGroup下所有在线客户端列表中。

当前消费对进行负载均衡策略后获取对应的消息消费队列。具体的算法很简单,可以看源码。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • java RocketMQ快速入门基础知识

    如何使用 1.引入 rocketmq-client <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.1.0-incubating</version> </dependency> 2.编写Producer DefaultMQProducer produce

  • Window搭建部署RocketMQ步骤详解

    序 以前简单用过ActiveMQ但是公司项目上使用的是RocketMQ,所以准备多花点时间在这上面,搞懂项目的配置使用. 看了很多资料,先说说我自己对RocketMQ的简单理解.不管是我们写的消费者还是生产者都属于客户端,而我们需要安装RocketMQ,这是属于服务端.和ActivieMQ.zookeeper类似,消费者.生成者.服务端(NameServer)之间是采取观察者模式实现. 在操作系统上安装RocketMQ,启动服务端NameServer.启动Broker,书写Consumer代码,

  • Docker中RocketMQ的安装与使用详解

    搜索RocketMQ的镜像,可以通过docker的hub.docker.com上进行搜索,也可以在Linux下通过docker的search命令进行搜索,不过最近防火墙升级后,导致国外的网站打开都很慢,通过命令搜索反而会更加方便,操作Docker命令一定要是root用户或者具有root权限的用户.查询操作如下: docker search rocketmq 可以得到如下的结果: 镜像倒是蛮多的,不过看来看去没有一个是官方发布的,我就随便选一个吧,如foxiswho/rocketmq,以下是一个查

  • springBoot整合RocketMQ及坑的示例代码

    版本: JDK:1.8 springBoot:1.5.10 rocketMQ:4.2.0 pom 配置: <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.10.RELEASE</version> </parent> <d

  • java rocketmq--消息的产生(普通消息)

    前言 与消息发送紧密相关的几行代码: 1. DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); 2. producer.start(); 3. Message msg = new Message(...) 4. SendResult sendResult = producer.send(msg); 5. producer.shutdown(); 那这几行代码执行时,背后都做了什么? 一. 首先

  • rocketmq消费负载均衡--push消费详解

    前言 本文介绍了DefaultMQPushConsumerImpl消费者,客户端负载均衡相关知识点.本文从DefaultMQPushConsumerImpl启动过程到实现负载均衡,从源代码一步一步分析,共分为6个部分进行介绍,其中第6个部分 rebalanceByTopic 为负载均衡的核心逻辑模块,具体过程运用了图文进行阐述. 介绍之前首先抛出几个问题: 1. 要做负载均衡,首先要解决的一个问题是什么? 2. 负载均衡是Client端处理还是Broker端处理? 个人理解: 1. 要做负载均衡

  • Windows下Apache+Tomcat7负载均衡配置方法详解

    准备工作 Windows Server 2008 R2 Enterprise(2.40GH,8GB,64Bit,192.168.10.212) 2台Ubuntu 10.04.4(192.168.10.98,192.168.10.137) JDK1.7.80 Tomcat7.0.68 Apache2.4.4 1.安装Apache2.4.4 安装过程中没什么注意事项,一直Next就行了.安装完成后默认开启Apache服务,可以在浏览器上输入localhost或者127.0.0.1,这时候会出现It

  • Nginx + consul + upsync 完成动态负载均衡的方法详解

    前置知识 DNS域名解析过程 在浏览器输入域名,访问后 在浏览器缓存中查找是否有对应的ip和端口,如果有直接访问对应ip和端口 浏览器缓存中没有则在本地host文件中查找是否有对应的~~ 本地host文件中没有则去DSN服务器上查找 外网映射 顾名思义,就是将本地的ip地址,映射出一个公网ip,可以供所有主机(连接到互联网的计算机)访问 疑问:为什么需要使用外网映射 例如做微信支付时的第三方接口回调,微信回调你的接口必须是一个公网ip地址,不然他根本无法给你回调,导致你本地无法测试.这个时候你就

  • CentOS6.5环境安装nginx服务器及负载均衡配置操作详解

    本文实例讲述了CentOS6.5环境安装nginx服务器及负载均衡配置操作.分享给大家供大家参考,具体如下: 1.下载PCRE, 是一个用C语言编写的正则表达式函数库 [root@localhost pcre-8.36]# cd /tmp/download/ [root@localhost download]# wget http://nchc.dl.sourceforge.net/project/pcre/pcre/8.37/pcre-8.37.tar.gz [root@localhost d

  • Django高并发负载均衡实现原理详解

    1 什么是负载均衡? 当一台服务器的性能达到极限时,我们可以使用服务器集群来提高网站的整体性能.那么,在服务器集群中,需要有一台服务器充当调度者的角色,用户的所有请求都会首先由它接收,调度者再根据每台服务器的负载情况将请求分配给某一台后端服务器去处理. 那么在这个过程中,调度者如何合理分配任务,保证所有后端服务器都将性能充分发挥,从而保持服务器集群的整体性能最优,这就是负载均衡问题. 下面详细介绍负载均衡的四种实现方式 2 HTTP重定向实现负载均衡 过程描述 当用户向服务器发起请求时,请求首先

  • Kafka消费客户端协调器GroupCoordinator详解

    目录 协调器的生命周期 GroupCoordinator的创建 offsetConfig相关配置 groupConfig相关配置 groupMetadataManager heartbeatPurgatory GroupCoordinator的启动 GroupCoordinator OnElection GroupCoordinator onResignation 协调器的生命周期 什么是协调器 协调器工作原理 协调器的Rebalance机制 GroupCoordinator的创建 在Kafka

  • RocketMQ事务消息原理与使用详解

    目录 一.RocketMQ事务消息概要 二.RocketMQ事务消息使用案例 (1).定义消息监听器 (2).定义消息生产者 (3).定义消息消费者 (4).观察生产者控制台输出 (5).观察消费者控制台输出 三.RocketMQ事务消息原理 四.RocketMQ事务消息使用限制 一.RocketMQ事务消息概要 RocketMQ事务消息(Transactional Message)是指应用本地事务和发送消息操作可以被定义到全局事务中,要么同时成功,要么同时失败.RocketMQ的事务消息提供类

  • MongoDB中的push操作详解(将文档插入到数组)

    目录 1. 概述 2. 数据库初始化 3. 使用 Mongo Query 进行推送操作 4. 使用Java驱动代码进行推送操作 4.1. 使用 DBObject 4.2. 使用 BSON 文档 5. 使用 addToSet操作符 5.1. 使用addToSet运算符的 Shell 查询 5.2. 使用addToSet运算符的 Java 驱动程序 6. 结论 总结 1. 概述 在本教程中,我们将介绍如何在MongoDB中将文档插入到数组中.此外,我们将看到 $push 和 $addToset 运算

  • vue拦截器Vue.http.interceptors.push使用详解

    刚开始学vue,github上down了一个开源项目,看源代码的时候看到了这个地方: /** * @export * @param {any} request * @param {any} next * @returns */ import store from './vuex/store' // 全局错误处理,全局loading import { setLoading, setTip } from './vuex/actions/doc_actions' export default func

  • RocketMq事务消息发送代码流程详解

    一.RocketMq事务消息流程: 1.首先会向broker发送一个预请求消息,消费者不可见 2.回调执行本地事务(比如操作数据库) 3.事务执行成功后,再次发送消息给broker,告诉broker事务执行成功这个消息要提交,让消费者可见.如果本地事务执行超时,会返回一个unknow,broker会发送一个消息回查,检查消息是否执行成功. 二.RocketMq事务消息实例: 1.引入rocketMq相关的依赖: <dependency> <groupId>org.apache.ro

随机推荐