深入研究RocketMQ消费者是如何获取消息的
前言
小伙伴们,国庆都过的开心吗?国庆后的第一个工作日是不是很多小伙伴还沉浸在假期的心情中,没有工作状态呢?
那王子今天和大家聊一聊RocketMQ的消费者是如何获取消息的,通过学习知识来找回状态吧。
废话不多说,我们开始吧。
消费者组
首先我们了解一个概念,什么是消费者组。
消费者组你就可以把它理解为,给一组消费者起一个名字。
假设我们有一个订单Topic名字是OrderTopic,然后库存系统和积分系统都要消费这个Topic中的数据,我们分别给库存系统和积分系统起一个消费组名字:stock_consumer_group、score_consumer_group。
设置消费者组名字是在代码中实现的,如下:
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("stock_consumer_group");
比如我们的库存系统提供了2台机器,每台机器上的消费者组名字都是stock_consumer_group,那么这2台机器就是一个消费者组。
大体结构如上图所示,那么当订单系统发送消息到OrderTopic中后,库存系统和积分系统是如何进行消费的呢?
默认情况下,这条消息发送到Broker后,库存系统和积分系统都会拉取这条消息,而且库存系统的两台机器中只有一台会消费到这条消息,积分系统也一样。
这就是消费组的概念,不同的系统设置不同的消费组,如果不同的消费组订阅了同一个Topic,那么对于Topic中的一条消息,每个消费组都会获取到这条消息。
集群模式和广播模式
接下来我们思考一个问题,对于消费者组而言,当它获取到一条消息后,假设消费者组内有多台机器,那么到底是只有一台机器获取到消息,还是所有机器都获取到消息呢?
这其实是消费的两种模式,集群模式和广播模式。
默认情况下我们都是使用的集群模式,也就是说消费者组收到消息后,只有其中的一台机器会接收到消息。
我们可以手动指定为广播模式。
consumer.setMessageModel(MessageModel.BROADCASTING)
指定为广播模式后,消费者组内的每台机器都会收到这条消息。
具体要根据业务场景选择消费模式。
MessageQueue与消费者的关系
接着我们想一下,对于一个Topic下的多个MessageQueue,消费者组中的多台机器是如何消费的呢?
这部分内容底层实现是很复杂的,我们可以简单的理解为它会均匀的将多个MessageQueue分配给消费者组中的多台机器消费。
举个例子,假如我们的OrderTopic有四个MessageQueue,这4个MessageQueue分布在两台MasterBroker上,每个MasterBroker上有两个MessageQueue。
然后库存系统作为一个消费者组有两台机器,那么最好的分配方式就是每台消费者机器负责两个MessageQueue,这样就实现了机器的负载消费,示意图如下:
所以我们可以大致的认为,一个Topic中的多个MessageQueue会被均匀的分布给一个消费者组中的多台机器进行消费,这里要注意一点,一个MessageQueue只能被一台消费者机器消费,但是一台消费者机器可以同时负责处理多个MessageQueue。
那么当消费者组中的机器数量发生变化时,是怎么处理的。
机器数量发生变化一般就两种情况,一种是有机器宕机了,另一种是增加机器进行集群扩容了。
其实这种情况下是会进行rebalance环节的,也就是会重新分配每个消费者机器要处理的MessageQueue。
Push模式和Pull模式
不知道小伙伴们还记不记得,在之前的文章RocketMQ的发送模式和消费模式中,我们已经用代码说明了消费者的两种消费模式:Push和Pull,当时只提供了Push消费的代码,而没有提供Pull消费的代码。
其实这两种模式本质上是一样的,都是消费者主动发出请求到Broker上拉取消息。
Push模式的底层也是通过消费者主动拉取的方式来实现的,只不过它的名字叫Push而已,意思是Broker尽可能实时的推送消息给消费者。
我们一般在使用RocketMQ的时候,消费模式基本都是使用的Push模式,因为Pull模式真的使用起来代码特别复杂,而且Push模式的底层还是Pull模式,只是对时效性有了更好的支持。
Push模式大体实现思路是这样的:当消费者发送请求到Broker拉取消息的时候,如果有新的消息可以消费,会立马返回消息到消费者进行消费,消费后会接着发送请求到Broker拉取消息。
也就说Push模式下,处理完一批消息后会理解再发送请求给Broker拉取下一批消息,所以时效性更好,看起来就像是Broker在实时推送消息。
当请求发送到Broker发现没有需要消费的消息时,就会让请求线程挂起,默认挂起15秒,然后会有另一个后台线程每隔一段时间判断一下是否有新消息需要消费,一旦发现了新的消息,就会去唤醒挂起的线程,将消息返回给消费者进行消费,然后消费完毕再次发送请求拉取消息。
这一部分的源码实现是很复杂的,我们只要了解它的核心思路就可以了。就算是Push模式,本质上也是对Pull模式的一种封装。
Broker如何读取消息返回给消费者
接下来我们来聊聊Broker是如何读取消息返回给消费者的。之前的文章深入研究Broker是如何持久化的中我们已经知道了Broker是如何持久化消息的,小伙伴们可以复习一下。
那么当消费者发送请求到Broker中拉取消息时,假设是第一次拉取,就会从MessageQueue中的第一条消息开始拉取。
如何定位到第一条消息的位置呢,首先Broker会找到MessageQueue对应的ConsumerQueue,从里面找到这条消息的offset,然后通过offset去CommitLog中读取消息数据,把消息返回给消费者。
当消费者消费完这条消息后,会提交一个消费的进度给Broker,Broker会记录下一个ConsumerOffset来标记我们的消费进度。
下次消费者再去这个MessageQueue中拉取消息时,就会从记录的消费位置继续拉取消息,而不用从头获取了。
总结
好了,到这里本篇文章就结束了。
今天主要和大家一起讨论了一下RocketMQ消费者的拉取和消费过程,也是国庆假期后的第一篇文章。
- Java基础-18(01)总结Map,HashMap,HashMap与Hashtable区别,Collections工具类
- 一个oracle查询引起的bug (r4笔记第59天)
- Java基础-18(02)总结Map,HashMap,HashMap与Hashtable区别,Collections工具类
- 特殊的物化视图刷新 (r4笔记第77天)
- 通过单例模式模拟RAC连接 (r4笔记第76天)
- 网站上的验证码是怎么产生的?
- Java基础-17(01)总结,登录注册案例,Set集合,HashSet
- mongoDB初探第一篇(r4笔记第75天)
- 重温快速排序(r4笔记第73天)
- Java基础-17(01)总结,TreeSet,LinkHashSet
- 海量数据迁移之数据抽取流程 (r4笔记第72天)
- CSS Selectors Level 4新特性全面解析
- 巧用外部表避免大量的insert (r4笔记第71天)
- 如何用java语言实现C#中的ref关键字(按引用传递参数)的效果
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- DC-1靶机实战和分析
- 如何用Python优雅的登录校园网?
- PHP入门之类型与运算符
- 栈论 : 递归与栈式访问,如何用栈实现所有递归操作(幼儿园题目篇,题目3)
- Maven是什么? Maven的概念+作用+仓库的介绍+常用命令
- JDK8;HashMap:再散列解决hash冲突 ,源码分析和分析思路
- 写一个 Singleton
- 树莓派基础实验31:MPU6050陀螺仪加速度传感器实验
- springboot gradle mybatis mysql配置(注解)
- PHP入门之流程控制
- 常用进制转换方法(取商留余)原理解析, 附基于栈实现进制转换的代码
- 用 jdom 解析 xml 文件时如何解决中文问题?如何解析?
- PHP入门之函数
- spring boot 启动报错 org/springframework/core/ErrorCoded
- PHP入门之数组