消息中间件之RocketMQ源码分析(三)

RocketMQ中的Consumer启动流程

RocketMQ客户端中有两个独立的消费者实现类分别为DefaultMQPullConsumer和DefaultMQPushConsumer，

DefaultMQPullConsumer

DefaultMQPullConsumer,该消费者使用时需要用户主动从Broker中Pull消息和消费消息，提交消费位点

继承关系图

在这里插入图片描述

核心属性

  • namesrvAddr:继承自ClientConfig,表示RocketMQ集群的Namesrv地址，如果是多个，则用逗号分开
    如:127.0.0.1:9876,127.0.0.2:9876
  • clientIP:使用客户端的程序所在机器的IP地址，目前支持IPV4和IPV6,同时排除了本地环会地址(127.0.xxx.xxx)
    和私有内网地址(192.168.xxx.xxx)，如果在Docket中运行，获取的IP地址是容器所在的IP地址，而非宿主主机的IP地址
  • instanceName:实例名，顾名思义每个实例都需要取不一样的名字，加入要在多个机器上部署
    多个程序进程，那么每个进程的实例名必须不相同，否则程序会启动失败，因为在创建MQClient时，
    会用到IP和instancename名称来
    在这里插入图片描述

在这里插入图片描述

  • vipChannelEnabled:这是一个boolean值，表示是否开启VIP通道。VIP通道和非VIP通道的区别是使用不同的端口号进行通信
  • clientCallbackExecutorThreads:客户端回调线程数。该线程数等于Netty通信层回调线程的个数，默认值为
    Runtime.getRuntime().availableProcessors();表示当前有效的CPU个数
  • pollNameServerInterval:获取Topic路由信息间隔，单位为ms，默认为30000ms(30s)
  • heartbeatBrokerInterval:客户端和Broker心跳间隔，单位为ms,默认30000ms(30s)
  • persistCOnsumerOffsetInterval:持久化消费位点时间间隔，单位为ms,默认为5000ms(5s)
  • defaultMQPullConsumer:默认pull消费者的具体实现
  • consumerGroup:消费者组名字
  • brokerSuspendMaxTimeMills:在长轮询模式下，Broker的最大挂起请求时间，建议不要修改此值
  • consumerTimeoutMillsWhenSuspend:在长轮询模式下，消费者的最大请求超时时间，必须比brokerSuspendMaxTimeMills大，不建议修改
  • messageModel:消费模式，现在支持集群模式消费和广播模式消费
  • messageQueueListener:消息路由信息变化时回调处理监听器，一般在重新平衡时被调用
  • offsetStore:位点存储模块。集群模式位点会持久化到Broker中，广播模式持久化到本地文件中(某个实例消费失败，生产者也不会重发)，位点存储模块有两个实现类，分别为RemoteBrokerOffsetStore和LocalFileOffsetStore
  • allocateMessageQUeueStrategy:消费Queue分配策略管理器，默认是平均分配策略
    private AllocateMessageQueueStrategy allocateMessageQueueStrategy = new AllocateMessageQueueAveragely();
  • maxReconsumeTimes:最大重试次数，可以配置

核心方法

  • registerMessageQueueListener():注册队列变化监听器，当队列发生变化是会被监听到
    在这里插入图片描述

  • pull():从Broker中Pull消息，如果有PullCallback参数，则表示异步拉取
    在这里插入图片描述

  • pullBlockIfNotFound():长轮询方式拉取，如果没有拉取到消息，那么Broker会讲请求Hold住一段时间，
    当有消息来临时再发送pull请求
    在这里插入图片描述

  • updateConsumeOffset():更新某一个Queue的消费位点
    在这里插入图片描述

  • fetchConsumeOffset():查找某个Queue的消费位点
    在这里插入图片描述

  • sendMessageBack():如果消费发送失败，则可以讲消息重新发回Broker，这个消费者组延迟一段时间后可以再消费(也就是重试)
    在这里插入图片描述

  • fetchSubscribeMessageQueues():获取一个Topic的全部Queue信息

在这里插入图片描述

Pull启动流程

在这里插入图片描述

  • 1.最初创建defaultMQPullConsumerImpl时的状态为ServiceState.CREATE_JSUT,然后设置消费者的默认启动状态为失败

在这里插入图片描述
在这里插入图片描述

  • 2.检查消费者的配置比，如消费者组名、消费类型、Queue的分配策略等参数是否符合规范，将订阅关系数据发给Rebalance服务对象
    在这里插入图片描述
    在这里插入图片描述

  • 3.校验消费者实例名，如果时默认的名字，则更改为当前的程序进程id
    在这里插入图片描述

  • 4.获取一个MQClientInstance,如果MQClientInstance已经初始化，则直接返回初始化的实例。这是核心对象，每个ClientID缓存一个实例
    在这里插入图片描述

  • 5.设置Rebalance对象消费组、消费类型、Queue分配策略、MQClientInstance等参数
    在这里插入图片描述

  • 6.对BrokerAPI的封装类pullAPIWrapper进行初始化，同时注册消息，过滤filter
    在这里插入图片描述

  • 7.初始化位点管理器并加载位点信息，位点管理器分为本地管理和远程管理，集群消费时
    消费位点保存在Broker中，由远程管理器管理，广播消息时位点在本地，由本地管理其管理
    在这里插入图片描述

  • 8.本地注册消费者实例，如果注册成功，则表示消费者启动成功
    在这里插入图片描述

DefaultMQPushConsumer

大部分属性、方法和DefaultMQPullConsumer是一样的

核心属性和方法

  • defaultMQPushConsumerImpl:默认的Push消费者具体实现类
  • consumeFromWhere:一个枚举，表示从什么位点开始消费，
    CONSUME_FROM_LAST_OFFSET：默认从上次消费的位点开始消费，相当于断点继续
    CONSUME_FROM_TIMESTAMP:从指定时间开始消费
    CONSUME_FROM_FIRST_OFFSET:从ConsumeQueue的最小位点开始消费
  • consumeTimestamp:表示从哪一时刻开始消费，时间格式为yyyyMMDDHHmmss,默认半小时前，当consumeFromWhere=CONSUME_FROM_TIMESTAMP时，consumeTimestamp设置的值才生效
  • allocateMessageQueueStrategy:消费者订阅topic-queue策略
  • subscription：订阅关系，表示当前消费者订阅了哪些Topic的哪些Tag
  • messageListener:消息Push回调监听器
  • consumeThreadMin:最小消费线程数，必须小于consumeThreadMax
    consumeThreadMax:最大线程数，必须大于consumeThreadMin
  • adjustThreadPoolNumsThreshold:动态调整消费线程池的线程数大小，开源版本不支持
  • consumeConcurrentlyMaxSpan:并发消息的最大位点差，，如果Pull消息的位点差超过该值，拉取变慢
  • pullThresholdForQueue:一个Queue能缓存的最大消息数，超过该值则采取拉取流控措施，默认是1000
  • pullThresholdSizeForQueue:一个Queue最大能缓存的消息字节数，单位是MB，默认是10MB
  • pullThresholdForTopic:一个Topic最大能缓存的消息数。超过该值则采取拉取流控措施，该字段值默认是-1，该值根据pullThresholdForQueue的配置决定是否生效，pullThresholdForTopic的优先级低于pullThresholdForQueue
  • pullThreasholdSizeForTopic:一个Topic最大能缓存的消息字节数，单位是MB,默认为-1,结合pullThresholdSizeForQueue配置项生效，该配置项的优先级低于pullThresholdSizeForQueue
  • pullInterval:拉取间隔，单位为ms
  • consumeMessageBatchMaxSize:消费者每次批量消费时，最多消费多少条消息，默认是1
  • pullBatchSize:一次最大拉取多少条消息，默认32条
  • postSubscriptionWhenPull:每次拉取消息时是否更新订阅关系，默认false
  • maxReconsumeTimes:最大重试次数，默认-1，表示最大重试次数为16次
  • suspendCurrentQueueTimeMillis为段轮询场景设置的挂起时间，比如顺序消息场景
  • consumeTimeout:消费超时时间，单位为min,默认是15

Push启动流程

在这里插入图片描述

  • 1-7和Pull模式类似
  • 8.初始化消费服务并启动，之所以用户"感觉"消息是Broker主动推送给自己的，
    是因为DefaultMQPushConsumer通过Pull服务将消息
    拉取到本地，再通过Callbakc的形式，将本地消息Push给用户的消费代码，
    DefaultMQPushConsumer和DefaultMQPullConsumer
    获取消息的方式一样，本质上都是拉取。

消费服务分为两种，即并行消费服务和顺序消费服务，对应的实现类分别是
ConsumeMessageConcurrentlyService和ConsumeMessageOrderlyService
根据用户监听器继承的不同接口初始化不同的消费服务程序

在这里插入图片描述

  • 9.启动MQClientInstance实例
    在这里插入图片描述
  • 10.更新本地订阅关系和路由信息，通过Broker检查是否支持消费者的过滤类型；
    向集群中的所有Broker发送消费者组的心跳信息
    在这里插入图片描述
  • 11.立即执行一次Rebalance
    this.mQClientFactory.rebalanceImmediately();

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/659050.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

vue-router 实现页面路由

vue-router 实现页面路由

vue-router介绍 vue 的官方路由组件 功能包括 嵌套路由映射动态路由选择模块化、基于组件的路由配置路由参数、查询、通配符HTML5 的 history 模式 和 hash 模式 vue-router使用 结合 tabs 组件，实现页面路由 安装 vant-ui 实现底部导航栏 Tabbar-CSDN博客 重点…
阅读更多...
图的学习

图的学习

图的基本概念和术语 图的定义：图是由顶点的有穷非空集合和顶点之间的边的集合组成的，G表示，V是图G中顶点的集合，E是图G中边的集合 无向图：任意两点的边都是无向边组成的图（无向边：&#xff08…
阅读更多...
R高级绘图 | P1 | 带边缘分布散点图 | 代码注释 + 结果解读

R高级绘图 | P1 | 带边缘分布散点图 | 代码注释 + 结果解读

新系列 —— R高级绘图，准备整理所有曾经绘制过的图图和未来需要的图图们的代码！预计这个系列会囊括所有常见图形，只提供高级绘图代码，基础绘图主要在 R语言绘图 系列中进行介绍，这个系列咱们主打：需要XX图…
阅读更多...
MySQL 常用函数学习总结

MySQL 常用函数学习总结

1、数值函数 ➢ 数学计算 ○ ABS(x) 返回x的绝对值 SELECT ABS(-2);○ EXP(x) 返回 e 的 x 次方 SELECT EXP(2); ○ LN 返回数字的自然对数 SELECT LN(2); ○ OG(x) 返回自然对数(以 e 为底的对数) SELECT LOG(20.085536923188); ○ POW(x,y)/POWER(x,y) 返回 x 的 y …
阅读更多...
Redis拒绝连接的原因与解决方式

Redis拒绝连接的原因与解决方式

Redis拒绝连接的原因与解决方式 在某些情况下，当尝试从外部计算机连接到运行在保护模式下的Redis服务器时，您可能会遇到如下的错误信息： Caused by: org.redisson.client.RedisException: DENIED Redis is running in protected mode becau…
阅读更多...
【Web前端实操21】商城官网_白色导航

【Web前端实操21】商城官网_白色导航

今日份实现白色导航栏部分，也就是第三部分，效果如图中划线所示： 本次实现代码如之前的全局样式不再赘述，如有需要可以去我博客的Web前端实操19或者20自行查看。 本次主要更新mi.css和index.htm。 实现导航栏所需要的CSS样…
阅读更多...
2015年苏州大学837复试机试C/C++

2015年苏州大学837复试机试C/C++

2015年苏州大学复试机试 第一题 题目 有36块砖&#xff0c;现在有36个人&#xff0c;男人能搬4块&#xff0c;女人能搬3块&#xff0c;小孩子两人搬一块&#xff0c;求一次搬完这些砖要男人&#xff0c;女人&#xff0c;小孩多少人&#xff1f; 代码 #include <iostrea…
阅读更多...
仰暮计划|“那时候在生产队下面，集体干活，吃大锅饭，由队里分粮食，吃不饱饭是常事，队里分的粮食就那么点，想要吃饱真的太难了”

仰暮计划|“那时候在生产队下面，集体干活，吃大锅饭，由队里分粮食，吃不饱饭是常事，队里分的粮食就那么点，想要吃饱真的太难了”

希望未来的中国越来越好&#xff0c;大家的生活也越来越好 老人是1955年在河南省洛阳市洛宁县的一个小山村里出生的&#xff0c;前半辈子为了生活&#xff0c;为了孩子而打拼&#xff0c;虽然经历了不少的苦难&#xff0c;但后半辈子也算是苦尽甘来&#xff0c;生活美满。现在就…
阅读更多...
LeetCode-1483. 树节点的第 K 个祖先

LeetCode-1483. 树节点的第 K 个祖先

原题链接&#xff1a;LeetCode-1483. 树节点的第 K 个祖先 题面 给你一棵树&#xff0c;树上有 n 个节点&#xff0c;按从 0 到 n-1 编号。树以父节点数组的形式给出&#xff0c;其中 parent[i] 是节点 i 的父节点。树的根节点是编号为 0 的节点。 树节点的第 k 个祖先节点是从…
阅读更多...
《Is dataset condensation a silver bullet for healthcare data sharing?》

《Is dataset condensation a silver bullet for healthcare data sharing?》

一篇数据浓缩在医疗数据集应用中的论文。 其实就是在医疗数据集上使用了data condensation的方法&#xff0c;这里使用了DM的方式&#xff0c;并且新增了浓缩时候使用不同的网络。 1. 方法 数据浓缩DC的目的是&#xff1a; E x ∼ P D [ L ( φ θ O ( x ) , y ) ] ≃ E x ∼…
阅读更多...
【Vue3+Vite】Vue3视图渲染技术 快速学习 第二期

【Vue3+Vite】Vue3视图渲染技术 快速学习 第二期

文章目录 一、模版语法1.1 插值表达式和文本渲染1.1.1 插值表达式 语法1.1.2 文本渲染 语法 1.2 Attribute属性渲染1.3 事件的绑定 二、响应式基础2.1 响应式需求案例2.2 响应式实现关键字ref2.3 响应式实现关键字reactive2.4 扩展响应式关键字toRefs 和 toRef 三、条件和列表渲…
阅读更多...
考研高数（共轭根式）

考研高数（共轭根式）

1.定义 共轭根式&#xff1a;是指两个不等于零的根式A、B&#xff0c;若它们的积AB不含根式&#xff0c;则称A、B互为共轭根式。 共轭根式的一个显著特点是通过相乘能把根号去掉&#xff0c;这是很有帮助的 2.常用的共轭根式 3.例题 1&#xff09;求极限 2&#xff09;证明…
阅读更多...
常见分类网络的结构

常见分类网络的结构

VGG16 图片来自这里 MobilenetV3 small和large版本参数,图片来着这里 Resnet 图片来自这里
阅读更多...
C++(8) 运算符重载

C++(8) 运算符重载

文章目录 运算符重载1.1 运算符概述1.2 运算符 重载1.3 运算符 << 重载 运算符重载 1.1 运算符概述 针对当前程序中常用的运算符&#xff0c;可以实现目标运算符处理当前数据类型的方式修改 需求&#xff1a; "ABCD" "BDC" > "ABCDBDC&qu…
阅读更多...
长度最小的子数组(c++题解)

长度最小的子数组(c++题解)

题目描述 给定一个含有 n 个正整数的数组和一个正整数 target 。 找出该数组中满足其和 ≥ target 的长度最小的 连续子数组 &#xff0c;并返回其长度。如果不存在符合条件的子数组&#xff0c;返回 0 。 输入格式 第一行输入两个整数n 和 target 第二行输入包含有n个元…
阅读更多...
【Deep Dive: AI Webinar】数据合作和开源人工智能

【Deep Dive: AI Webinar】数据合作和开源人工智能

【深入探讨人工智能】网络研讨系列总共有 17 个视频。我们按照视频内容&#xff0c;大致上分成了 3 个大类&#xff1a; 1. 人工智能的开放、风险与挑战&#xff08;4 篇&#xff09; 2. 人工智能的治理&#xff08;总共 12 篇&#xff09;&#xff0c;其中分成了几个子类&…
阅读更多...
课程表分析

课程表分析

课程表分析 class Course(models.Model):name models.CharField(max_length64)title models.CharField(max_length64)students models.IntegerField(default0)level models.IntegerField(choices((0, 入门), (1, 进阶)), default0)time models.IntegerField(default0)det…
阅读更多...
02、全文检索 ------ Solr（企业级的开源的搜索引擎） 的下载、安装、Solr的Web图形界面介绍

02、全文检索 ------ Solr（企业级的开源的搜索引擎） 的下载、安装、Solr的Web图形界面介绍

目录 Solr 的下载和安装Solr的优势&#xff1a;Lucene与Solr 安装 Solr1、下载解压2、添加环境变量3、启动 Solr Solr 所支持的子命令&#xff1a;Solr 的 Core 和 Collection 介绍Solr 的Web控制台DashBoard&#xff08;仪表盘&#xff09;Logging&#xff08;日志&#xff09…
阅读更多...
代码随想录算法训练营29期|day34 任务以及具体任务

代码随想录算法训练营29期|day34 任务以及具体任务

第八章 贪心算法 part03 1005.K次取反后最大化的数组和 class Solution {public int largestSumAfterKNegations(int[] nums, int K) {// 将数组按照绝对值大小从大到小排序&#xff0c;注意要按照绝对值的大小nums IntStream.of(nums).boxed().sorted((o1, o2) -> Math.ab…
阅读更多...
华为1.24秋招笔试题

华为1.24秋招笔试题

华为1.24秋招笔试题 1.题目1 题目详情 - 2024.1.24-华为秋招笔试-第一题-计算积分 - CodeFun2000 1.1题解 import java.util.Scanner;class Main{public static void main(String[] args){Scanner scnew Scanner(System.in);String ssc.next();char[] chs.toCharArray();in…
阅读更多...
最新文章

Copyright @ 2022~2023