当前位置: 首页 > news >正文

RocketMQ源码 Broker-SubscriptionGroupManager 订阅组管理组件源码分析

前言

SubscriptionGroupManager 继承了ConfigManager配置管理组件,拥有将内存数据持久化到磁盘文件subscriptionGroup.json的能力。它主要负责维护所有消费组在内存中的订阅数据。


源码版本:4.9.3

源码架构图

核心数据结构

主要的数据结构比较简单,维护了Map<消费组名称, 订阅组配置>的映射关系。

// 订阅组管理组件
public class SubscriptionGroupManager extends ConfigManager {private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);// Map<消费组名称,订阅组配置>private final ConcurrentMap<String, SubscriptionGroupConfig> subscriptionGroupTable =new ConcurrentHashMap<String, SubscriptionGroupConfig>(1024);// 内存数据版本号private final DataVersion dataVersion = new DataVersion();
}

深入看下SubscriptionGroupConfig 的数据结构。

public class SubscriptionGroupConfig {// 消费组名称private String groupName;// 是否开启消费private boolean consumeEnable = true;// 是否允许消费最早消息private boolean consumeFromMinEnable = true;// 是否允许广播消费private boolean consumeBroadcastEnable = true;// 重试队列数private int retryQueueNums = 1;// 重试最大次数private int retryMaxTimes = 16;// brokerIdprivate long brokerId = MixAll.MASTER_ID;// 当产生慢消费时,选择第几个brokerprivate long whichBrokerWhenConsumeSlowly = 1;// 是否通知消费者ids变化private boolean notifyConsumerIdsChangedEnable = true;
}

核心数据行为

数据行为主要都是对上面提到的数据结构的维护,代码 + 注释如下:

// 订阅组管理组件
public class SubscriptionGroupManager extends ConfigManager {public SubscriptionGroupManager() {this.init();}public SubscriptionGroupManager(BrokerController brokerController) {this.brokerController = brokerController;this.init();}private void init() {{// 初始化系统消费组SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.TOOLS_CONSUMER_GROUP);this.subscriptionGroupTable.put(MixAll.TOOLS_CONSUMER_GROUP, subscriptionGroupConfig);}{// 初始化过滤服务消费组SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.FILTERSRV_CONSUMER_GROUP);this.subscriptionGroupTable.put(MixAll.FILTERSRV_CONSUMER_GROUP, subscriptionGroupConfig);}{// 初始化自测消费组SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.SELF_TEST_CONSUMER_GROUP);this.subscriptionGroupTable.put(MixAll.SELF_TEST_CONSUMER_GROUP, subscriptionGroupConfig);}{// 初始化http代理消费组SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.ONS_HTTP_PROXY_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true);this.subscriptionGroupTable.put(MixAll.ONS_HTTP_PROXY_GROUP, subscriptionGroupConfig);}{// 初始化ONS_API_PULL消费组SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.CID_ONSAPI_PULL_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true); // 激活广播模式this.subscriptionGroupTable.put(MixAll.CID_ONSAPI_PULL_GROUP, subscriptionGroupConfig);}{// 初始化ONS_API_PERMISSION消费组SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.CID_ONSAPI_PERMISSION_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true);this.subscriptionGroupTable.put(MixAll.CID_ONSAPI_PERMISSION_GROUP, subscriptionGroupConfig);}{// 初始化ONS_API_OWNER消费组SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.CID_ONSAPI_OWNER_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true);this.subscriptionGroupTable.put(MixAll.CID_ONSAPI_OWNER_GROUP, subscriptionGroupConfig);}}// 更新订阅配置,且更新内存数据版本号public void updateSubscriptionGroupConfig(final SubscriptionGroupConfig config) {SubscriptionGroupConfig old = this.subscriptionGroupTable.put(config.getGroupName(), config);if (old != null) {log.info("update subscription group config, old: {} new: {}", old, config);} else {log.info("create new subscription group, {}", config);}this.dataVersion.nextVersion();this.persist();}// 失效消费组public void disableConsume(final String groupName) {SubscriptionGroupConfig old = this.subscriptionGroupTable.get(groupName);if (old != null) {old.setConsumeEnable(false);this.dataVersion.nextVersion();}}// 查找指定消费组的订阅配置public SubscriptionGroupConfig findSubscriptionGroupConfig(final String group) {SubscriptionGroupConfig subscriptionGroupConfig = this.subscriptionGroupTable.get(group);if (null == subscriptionGroupConfig) {if (brokerController.getBrokerConfig().isAutoCreateSubscriptionGroup() || MixAll.isSysConsumerGroup(group)) {subscriptionGroupConfig = new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(group);SubscriptionGroupConfig preConfig = this.subscriptionGroupTable.putIfAbsent(group, subscriptionGroupConfig);if (null == preConfig) {log.info("auto create a subscription group, {}", subscriptionGroupConfig.toString());}this.dataVersion.nextVersion();this.persist();}}return subscriptionGroupConfig;}// 将内存数据结构编码成字符串@Overridepublic String encode() {return this.encode(false);}// 获取配置文件路径@Overridepublic String configFilePath() {return BrokerPathConfigHelper.getSubscriptionGroupPath(this.brokerController.getMessageStoreConfig().getStorePathRootDir());}// 从字符串中恢复数据,写回内存数据结构@Overridepublic void decode(String jsonString) {if (jsonString != null) {SubscriptionGroupManager obj = RemotingSerializable.fromJson(jsonString, SubscriptionGroupManager.class);if (obj != null) {this.subscriptionGroupTable.putAll(obj.subscriptionGroupTable);this.dataVersion.assignNewOne(obj.dataVersion);this.printLoadDataWhenFirstBoot(obj);}}}// 将内存数据结构编码成字符串public String encode(final boolean prettyFormat) {return RemotingSerializable.toJson(this, prettyFormat);}// 当第一次启动时,打印加载数据时的日志private void printLoadDataWhenFirstBoot(final SubscriptionGroupManager sgm) {Iterator<Entry<String, SubscriptionGroupConfig>> it = sgm.getSubscriptionGroupTable().entrySet().iterator();while (it.hasNext()) {Entry<String, SubscriptionGroupConfig> next = it.next();log.info("load exist subscription group, {}", next.getValue().toString());}}public ConcurrentMap<String, SubscriptionGroupConfig> getSubscriptionGroupTable() {return subscriptionGroupTable;}public DataVersion getDataVersion() {return dataVersion;}// 删除指定消费组的订阅配置public void deleteSubscriptionGroupConfig(final String groupName) {SubscriptionGroupConfig old = this.subscriptionGroupTable.remove(groupName);if (old != null) {log.info("delete subscription group OK, subscription group:{}", old);this.dataVersion.nextVersion();this.persist();} else {log.warn("delete subscription group failed, subscription groupName: {} not exist", groupName);}}
}

相关文章:

RocketMQ源码 Broker-SubscriptionGroupManager 订阅组管理组件源码分析

前言 SubscriptionGroupManager 继承了ConfigManager配置管理组件&#xff0c;拥有将内存数据持久化到磁盘文件subscriptionGroup.json的能力。它主要负责维护所有消费组在内存中的订阅数据。 源码版本&#xff1a;4.9.3 源码架构图 核心数据结构 主要的数据结构比较简单&am…...

go-zero开发入门-API网关鉴权开发示例

本文是go-zero开发入门-API网关开发示例一文的延伸&#xff0c;继续之前请先阅读此文。 在项目根目录下创建子目录 middleware&#xff0c;在此目录下创建文件 auth.go&#xff0c;内容如下&#xff1a; // 鉴权中间件 package middlewareimport ("context""e…...

[LLM]nanoGPT---训练一个写唐诗的GPT

karpathy/nanoGPT: The simplest, fastest repository for training/finetuning medium-sized GPTs. (github.com) 原有模型使用的莎士比亚的戏剧数据集, 如果需要一个写唐诗机器人&#xff0c;需要使用唐诗的文本数据&#xff0c; 一个不错的唐诗&#xff0c;宋词数据的下载…...

docker compose部署wordpress

准备机器&#xff1a; 192.168.58.151 &#xff08;关闭防火墙和selinux&#xff09; 安装好docker服务 &#xff08;详细参照&#xff1a;http://t.csdnimg.cn/usG0s 中的国内源安装docker&#xff09; 部署wordpress: 创建目录&#xff1a; [rootdocker ~]# mkdir…...

【docker四】使用Docker-compose一键部署Wordpress平台

目录 一、YAML 文件格式及编写注意事项&#xff08;重要&#xff09; 1、yaml文件使用时注意事项&#xff1a; 2、yaml文件的基本数据结构&#xff1a; 2.1、声明变量&#xff08;标量。是单个的不可再分的值&#xff0c;类型&#xff1a;字符串&#xff0c;整数&#xff0c…...

HTML程序大全(1):简易计算器

HTML代码&#xff0c;主要创建了几个按钮。 <div class"container"><div class"output" id"output">0</div><button class"button" onclick"clearOutput()" id"clear">C</button>…...

esp32服务器与android客户端的tcp通讯

esp32 //esp32作为服务端 #include <WiFi.h>#define LED_BUILTIN 2 // 创建热点 const char *ssid "ESP32"; const char *password "12345678"; const int port 1122; //端口 WiFiServer server(port); void setup() {delay(5000);pinMode(LED_…...

自定义Mybatis LanguageDriver性能优化

场景&#xff1a;高并发情况下mybatis 动态sql 解析 锁问题优化 优化前 并发测试 XMLLanguageDriver 类 的 createSqlSource 方法有锁 而且 每次执行时都会走该方法 优化前 &#xff1a; 线程有Block 优化后的 LanguageDriver public class CustomXMLLanguageDriver im…...

DevEco Studio 鸿蒙(HarmonyOS)项目结构

DevEco Studio 鸿蒙&#xff08;HarmonyOS&#xff09;项目结构 一、操作环境 操作系统: Windows 10 专业版 IDE:DevEco Studio 3.1 SDK:HarmonyOS 3.1 二、项目结构 创建简单的Hello World移动应用项目结构如下图 由上到下说明各个文件夹的作用 .hvigor&#xff1a;存…...

Springboot整合篇Druid

一、概述 1.1简介 Druid 是阿里巴巴开源平台上一个数据库连接池实现&#xff0c;结合了 C3P0、DBCP 等 DB 池的优点&#xff0c;同时加入了日志监控。 它本身还自带一个监控平台&#xff0c;可以查看时时产生的sql、uri等监控数据&#xff0c;可以排查慢sql、慢请求&#xff0…...

uniapp 微信小程序 封装axios 包含请求拦截、响应拦截、无感刷新令牌功能

前言&#xff1a; 1、为什么不适用uniapp自带的请求功能&#xff1f; 答&#xff1a;uniapp自带的请求功能&#xff0c;再刷新了令牌后&#xff0c;重新请求返回的数据无法返回给发起请求的方法。也就是说&#xff0c;刷新令牌后重新发起的请求和第一次发起请求的方法是割裂的。…...

C语言精选——选择题Day41

第一题 1. 有以下程序段&#xff1a; char *p, *q; p (char *)malloc(sizeof(char) * 20); q p; scanf("%s %s", p, q); printf("%s %s\n", p, q); 若从键盘输入&#xff1a;abc def↙&#xff0c;则输出结果是&#xff08; &#xff09; A&#xff1a;d…...

Tomcat头上有个叉叉

问题原因&#xff1a; 这是因为它就是个空的tomcat,并没有导入项目运行 解决方案&#xff1a; war模式&#xff1a;发布模式&#xff0c;正式发布时用&#xff0c;将WEB工程以war包的形式上传到服务器 war exploded模式&#xff1a;开发时用&#xff0c;将WEB工程的文件夹直接…...

Linux shell编程学习笔记35:seq

0 前言 在使用 for 循环语句时&#xff0c;我们经常使用到序列。比如&#xff1a; for i in 1 2 3 4 5 6 7 8 9 10; do echo "$i * 2 $(expr $i \* 2)"; done 其中的 1 2 3 4 5 6 7 8 9 10;就是一个整数序列 。 为了方便我们使用数字序列&#xff0c;Linux提供了…...

Nougat:结合光学神经网络,引领学术PDF文档的智能解析、挖掘学术论文PDF的价值

Nougat&#xff1a;结合光学神经网络&#xff0c;引领学术PDF文档的智能解析、挖掘学术论文PDF的价值 这是Nougat的官方存储库&#xff0c;Nougat是一种学术文档PDF解析器&#xff0c;可以理解LaTeX数学和表格。 Project page: https://facebookresearch.github.io/nougat/ …...

涉密网络的IP查询防护策略

涉密网络的安全性对于维护国家、企业及个人的核心利益至关重要。在当今数字化时代&#xff0c;网络攻击日益猖獗&#xff0c;其中IP查询是攻击者获取目标信息的一种常见手段。本文将探讨涉密网络中防护IP查询的关键策略&#xff0c;以确保网络的机密性和安全性。 1. 专用VPN和…...

基础算法(1):排序(1):选择排序

今天对算法产生了兴趣&#xff0c;开始学习基础算法&#xff0c;比如排序&#xff0c;模拟&#xff0c;贪心&#xff0c;递推等内容&#xff0c;算法是很重要的&#xff0c;它是解决某个问题的特定方法&#xff0c;程序数据结构算法&#xff0c;所以对算法的学习是至关重要的&a…...

GeoTrust OV证书

当谈到网站安全性和可信度时&#xff0c;GeoTrust OV证书是一个备受推崇的选择。作为一家备受尊敬的数字证书颁发机构&#xff0c;GeoTrust以其卓越的品牌声誉和高质量的产品而闻名于世。GeoTrust OV证书提供了一系列的安全功能&#xff0c;同时还具有出色的性价比&#xff0c;…...

第一个“hello Android”程序

1、首先安装Android studio&#xff08;跳过&#xff09; Android Studio是由Google推出的官方集成开发环境&#xff08;IDE&#xff09;&#xff0c;专门用于Android应用程序的开发。它是基于JetBrains的IntelliJ IDEA IDE构建的&#xff0c;提供了丰富的功能和工具&#xff0…...

docker-compose安装nacos和msql

docker-compose安装nacos和msql 前言前提已经安装docker-compose&#xff0c;如果没有安装&#xff0c;则可以查看上面系列文章中的安装教程。并且文章中使用的是mobaxterm连接虚拟机。 1、下载2、创建并运行 前言 前提已经安装docker-compose&#xff0c;如果没有安装&#x…...

第19节 Node.js Express 框架

Express 是一个为Node.js设计的web开发框架&#xff0c;它基于nodejs平台。 Express 简介 Express是一个简洁而灵活的node.js Web应用框架, 提供了一系列强大特性帮助你创建各种Web应用&#xff0c;和丰富的HTTP工具。 使用Express可以快速地搭建一个完整功能的网站。 Expre…...

基于大模型的 UI 自动化系统

基于大模型的 UI 自动化系统 下面是一个完整的 Python 系统,利用大模型实现智能 UI 自动化,结合计算机视觉和自然语言处理技术,实现"看屏操作"的能力。 系统架构设计 #mermaid-svg-2gn2GRvh5WCP2ktF {font-family:"trebuchet ms",verdana,arial,sans-…...

智慧工地云平台源码,基于微服务架构+Java+Spring Cloud +UniApp +MySql

智慧工地管理云平台系统&#xff0c;智慧工地全套源码&#xff0c;java版智慧工地源码&#xff0c;支持PC端、大屏端、移动端。 智慧工地聚焦建筑行业的市场需求&#xff0c;提供“平台网络终端”的整体解决方案&#xff0c;提供劳务管理、视频管理、智能监测、绿色施工、安全管…...

vscode(仍待补充)

写于2025 6.9 主包将加入vscode这个更权威的圈子 vscode的基本使用 侧边栏 vscode还能连接ssh&#xff1f; debug时使用的launch文件 1.task.json {"tasks": [{"type": "cppbuild","label": "C/C: gcc.exe 生成活动文件"…...

【SQL学习笔记1】增删改查+多表连接全解析(内附SQL免费在线练习工具)

可以使用Sqliteviz这个网站免费编写sql语句&#xff0c;它能够让用户直接在浏览器内练习SQL的语法&#xff0c;不需要安装任何软件。 链接如下&#xff1a; sqliteviz 注意&#xff1a; 在转写SQL语法时&#xff0c;关键字之间有一个特定的顺序&#xff0c;这个顺序会影响到…...

苍穹外卖--缓存菜品

1.问题说明 用户端小程序展示的菜品数据都是通过查询数据库获得&#xff0c;如果用户端访问量比较大&#xff0c;数据库访问压力随之增大 2.实现思路 通过Redis来缓存菜品数据&#xff0c;减少数据库查询操作。 缓存逻辑分析&#xff1a; ①每个分类下的菜品保持一份缓存数据…...

解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错

出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上&#xff0c;所以报错&#xff0c;到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本&#xff0c;cu、torch、cp 的版本一定要对…...

Map相关知识

数据结构 二叉树 二叉树&#xff0c;顾名思义&#xff0c;每个节点最多有两个“叉”&#xff0c;也就是两个子节点&#xff0c;分别是左子 节点和右子节点。不过&#xff0c;二叉树并不要求每个节点都有两个子节点&#xff0c;有的节点只 有左子节点&#xff0c;有的节点只有…...

AI书签管理工具开发全记录(十九):嵌入资源处理

1.前言 &#x1f4dd; 在上一篇文章中&#xff0c;我们完成了书签的导入导出功能。本篇文章我们研究如何处理嵌入资源&#xff0c;方便后续将资源打包到一个可执行文件中。 2.embed介绍 &#x1f3af; Go 1.16 引入了革命性的 embed 包&#xff0c;彻底改变了静态资源管理的…...

GC1808高性能24位立体声音频ADC芯片解析

1. 芯片概述 GC1808是一款24位立体声音频模数转换器&#xff08;ADC&#xff09;&#xff0c;支持8kHz~96kHz采样率&#xff0c;集成Δ-Σ调制器、数字抗混叠滤波器和高通滤波器&#xff0c;适用于高保真音频采集场景。 2. 核心特性 高精度&#xff1a;24位分辨率&#xff0c…...