RocketMQ-订阅一致及解决方案

背景

这里借用Rocketmq官方的一句话来描述订阅关系一致:

订阅关系一致指的是同一个消费者分组Group ID下，所有Consumer实例所订阅的Topic和Tag必须完全一致。如果订阅关系不一致，可能导致消息消费逻辑混乱，消息被重复消费或遗漏。

具体的问题和实例请看阿里云关于Rocketmq订阅关系一致的说明 ,里面写的非常详细,这边主要是讨论一下关于经典的会出现的一个订阅不一致问题。

当前问题

我司由于历史问题，java侧服务mq使用泛滥，每多一个topic订阅就伴随着新建一个group，导致维护成本越来越高，所以我们在2.0 sdk第一版即支持 【一个消费group消费多个topic】，也就是如下面这张图的预期：
在这里插入图片描述
看起来没有问题,RocketMQ官方也支持多topic的订阅逻辑，我们也是这么去推动大家升级的。但是随着对MQ的深入了解，逐渐发现一个很可怕的问题： 如果一个正在使用的group我希望去对它进行订阅关系的变更(添加/删除topic订阅)，这个是绝对没有办法走灰度发布的！因为它会直接出现

RocketMQ领域经典的订阅不一致问题，详情见下图(模拟了一个使用中的group变更订阅关系时的灰度发布过程)
在这里插入图片描述
由图中可知,当前sdk虽然支持了一个group监听多个topic，但是这仅限于新业务，一个全新的group才可以在一开始用这种方式去升级，但却没有办法支持后续的订阅关系变更，看起来之前的sdk升级没什么用，可扩展性太差。如果消息的收发都是新业务还好一点，假如是订阅一个发送量非常大的现有topic，一发版就会喜提告警，严重的会存在消息丢失的风险，并且无法回放。

解决方案

其实问题的关键在于： 每个客户端虽然知道其他客户端的存在，但是并不知道大家的订阅关系，就导致了在实际平衡的时候产生【我觉得他应该去消费这些队列】的错觉，所以解决问题的关键就是我们只要让每个客户端都知道整个group集群中所有客户端的订阅关系就行了。参考之前发表的rocketmq灰度方案，可以利用ClientId的特性，将当前客户端的订阅关系加密追加在ID后面。

public String buildMQClientId() {StringBuilder sb = new StringBuilder();sb.append(this.getClientIP());sb.append("@");sb.append(this.getInstanceName());if (!UtilAll.isBlank(this.unitName)) {sb.append("@");sb.append(this.unitName);}if (enableStreamRequestType) {sb.append("@");sb.append(RequestType.STREAM);}# 关键在于下面这几行代码MessageInstance instance = MessageStorage.getInstance(this.getInstanceName());if (instance != null) {sb.append("#");sb.append(MessageStorage.generateInstanceSubInfoEncode(instance));} else {sb.append("#[]");}return sb.toString();}

关于instance、group、topic的关系可以看下面这张图:

在这里插入图片描述

每个服务进程使用binder可以收发不同实例下的消息，因此在SDK中ClientId是以订阅的实例为维度创建的，在RocketMQ源码中是单例模式。

然后可以自己实现一个负载均衡策略:

/*** 消息队列分配策略增强--保证不出现订阅不一致的情况** @author mobai* @since 2024/6/9 12:57 AM*/
@Slf4j
public class EnhanceAllocateMessageQueueStrategyImpl extends AllocateMessageQueueAveragely {/*** 保证订阅一致的分配算法* 如果有任意客户端sdk版本低于当前版本,则降为默认的平均分配算法* <p> 1.如果是重试topic，则使用平均分配策略(重试的topic走的是内部回传broker,写到哪一个队列是随机的)* <p> 2.通过clientId获取每个client的订阅信息,然后获取客户端中对应当前group的topic监听列表，判断当前需要平衡的topic是否在监听列表中,* 如果不在则认为订阅不一致，让所有订阅了当前topic的客户端去分配所有的队列* <p> 3.如果订阅一致，则使用平均分配策略* 同时提供了一个允许覆盖的分配方法,默认是平均分配。子类可以根据实际情况自行覆盖，该方法会传入当前的订阅是否不出现不一致** @param consumerGroup 当前消费者组* @param currentCID    当前客户端id* @param mqAll         当前topic下的所有队列* @param cidAll        当前group的云端所有客户端实例* @return 分配结果*/@Overridepublic List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {if (!check(consumerGroup, currentCID, mqAll, cidAll)) {return Collections.emptyList();}if (mqAll.stream().anyMatch(mq -> mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX))) {return super.allocate(consumerGroup, currentCID, mqAll, cidAll);}String topic = mqAll.get(0).getTopic();boolean isSomeClientVersionLower = cidAll.stream().anyMatch(c -> c.lastIndexOf(MqConstant.GROUP_ENHANCE_TAG) == -1);if (isSomeClientVersionLower) {//避免当前这个增强sdk版本在灰度的时候，出现低版本客户端log.warn("[enhance allocate]: group:{}sub topic:{} has lower version client,use the default avg strategy", consumerGroup, topic);return super.allocate(consumerGroup, currentCID, mqAll, cidAll);}if (log.isDebugEnabled()) {log.info("[enhance allocate]: group:{} start topic rebalance:{},current client num:{},current queues num:{}", consumerGroup, topic, cidAll.size(), mqAll.size());}Map<String, List<MessageConsumer>> allClientsSubInfo = MessageStorage.getDecodeSubInfo(cidAll);Map<String, MessageConsumer> eachClientGroup = new HashMap<>(allClientsSubInfo.size());allClientsSubInfo.forEach((k, v) -> {for (MessageConsumer messageConsumer : v) {if (messageConsumer.getActualGroup().equals(consumerGroup)) {eachClientGroup.put(k, messageConsumer);break;}}});List<String> validCids = new ArrayList<>(eachClientGroup.size());for (Map.Entry<String, MessageConsumer> consumerEntry : eachClientGroup.entrySet()) {List<MessageConsumer.ListenTopic> currentConsumerSubTopics = consumerEntry.getValue().getTopics();if (currentConsumerSubTopics.stream().anyMatch(listenTopic ->listenTopic.getActualTopic().equals(topic)|| listenTopic.getTopic().equals(topic)|| listenTopic.getSourceTopic().equals(topic))) {validCids.add(consumerEntry.getKey());}}//如果存在订阅不一致的情况,则让所有订阅了当前topic的客户端去分配所有的队列,并且此逻辑不允许扩展,优先保证消息安全不丢失、不堆积if (validCids.size() != cidAll.size()) {List<MessageQueue> messageQueues = balanceAllocate(consumerGroup, currentCID, mqAll, validCids);log.warn("[enhance allocate]: group:{}sub topic:{} has not-balance-sub condition，sdk start enhance,clients {} complete {} queues rebalance,currentId:{},\n allocate result:{}", consumerGroup, topic,MessageStorage.getClientsIp(validCids), mqAll.size(), currentCID, MessageStorage.joinMessageQueue(messageQueues));return messageQueues;} else {return doAllocate(consumerGroup, currentCID, mqAll, cidAll);}}/*** 可扩展的分配算法,默认是平均分配** @param consumerGroup 消费组* @param currentCID    当前消费者* @param mqAll         所有消息队列* @param cidAll        所有消费者* @param isSubBalance  是否订阅均衡* @return 分配结果*/public List<MessageQueue> doAllocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {return balanceAllocate(consumerGroup, currentCID, mqAll, cidAll);}/*** 平均分配算法** @param consumerGroup 消费组* @param currentCID    当前消费者* @param mqAll         所有消息队列* @param cidAll        所有消费者* @return 消息队列*/public final List<MessageQueue> balanceAllocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {return super.allocate(consumerGroup, currentCID, mqAll, cidAll);}@Overridepublic String getName() {return "Enhance";}

策略继承于平均分配策略,大概的思路如下：

  1. 排除掉重试topic
  2. 通过clientId判断是否存在不同版本的SDK，这点也很重要，当这个增强的策略在发布时，因为线上的服务并没有该ClientId标识，所以此时退化成标准的平均分配是最安全的。
  3. 通过将所有客户端Id进行信息提取和解密，判断当前balance的topic有哪一些客户端在监听(当前group肯定会监听,不然这个方法链路进不来)
  4. 如果发现过滤出来的客户端个数和云上记录的所有客户端个数不同，即认定为订阅不一致，此时让有当前topic订阅关系的客户端分配所有队列，这个逻辑禁止覆盖
  5. 在保证订阅一致的前提下,提供了一个允许扩展的分配算法，默认使用平均分配(灰度消息就是通过继承此类，扩展该方法实现的保证一致性的前提下做的灰度)
  6. 那些没有订阅当前topic的客户端进程不会进到这个topic的平衡方法

升级了SDK之后，以下是对应的交互变更效果图(只讨论新增订阅关系的场景,删除订阅关系也是一个道理)

在这里插入图片描述

验证

接下来通过一个服务来验证此逻辑的可行性(包含了灰度消息逻辑)，首先准备了一个订阅了一个topic的group，sdk版本是2.0.8(没有该增强逻辑)

已知：topic有64个队列,存在8个broker上，消费已做好幂等。
在这里插入图片描述

升级该服务的sdk版本到2.1.0(当前增强版本),订阅关系不变，发布灰度

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

sdk判断当前客户端存在版本不一致，因此降级为默认平均分配算法,发送10条消息测试一下

在这里插入图片描述
消费正常。

升级该服务SDK到2.1.0，直接发布上线，无订阅关系变更

在这里插入图片描述
在这里插入图片描述
队列分配正确，再发送10条消息：

在这里插入图片描述
消费正常。

新增加一个topic的订阅关系,发布灰度(新topic48个队列，分布在6个brokder上)

控制台提示订阅不一致

在这里插入图片描述
灰度pod日志: 独自接管了新topic全部队列，旧topic获取到每个brokder最后一个分区

在这里插入图片描述
在这里插入图片描述
正常pod日志:不受影响，只和消费之前的topic(灰度pod消费每个broker最后一个分区),所以只分配到到56个队列

在这里插入图片描述
在这里插入图片描述
此时发送10条消息到新的topic上，结果消息全部被灰度也就是新加订阅关系的客户端全部消费

在这里插入图片描述
再发送10条老消息到旧topic上，9条在正常的pod，1条在灰度的pod，也符合灰度只负载1/10分区的策略

在这里插入图片描述
验证通过,灰度验证通过

在这里插入图片描述
订阅一致了

减少其中一个topic的订阅关系,再次发布灰度

控制台订阅不一致

在这里插入图片描述
灰度pod(减少订阅的客户端)日志：只参与旧topic的分配,且是灰度分区，其他无影响

在这里插入图片描述
在这里插入图片描述
正常pod(完整订阅关系)日志: 新topic提示不一致，进入增强逻辑,分配到全部的48个队列，旧topic分配正常

在这里插入图片描述
在这里插入图片描述
发送10条消息到被删除订阅关系的新topic: 全部被有订阅关系的正常客户端消费

在这里插入图片描述
发送10条消息到老的共有的老topic: 9比1的比例被俩客户端平均消费

在这里插入图片描述
验证通过。

结论

该方案被验证是安全可行的，但是在实际接入时需要注意:

  1. 不要在首次升级sdk时就变更订阅关系发灰度,这样的话还是会出现订阅不一致，无解，一个比较好的做法是先将SDK版本全部升级(允许灰度)，等后续版本迭代再做订阅关系的变更，就可以正常发灰度验证。
  2. 生产环境永远不要使用公网接入点,除了安全问题之外，阿里云公网接入点架构模式是服务端负载，该策略会失效，而且原则上生产也不应该开放公网接入点。

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

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

相关文章

代码随想录第44天|动态规划

代码随想录第44天|动态规划

188.买卖股票的最佳时机IV 309.最佳买卖股票时机含冷冻期 参考 class Solution { public:int maxProfit(vector<int>& prices) {vector<vector<int>> dp(prices.size(), vector<int>(4, 0));dp[0][0] -prices[0];dp[0][1] 0;dp[0][2] 0;d…
阅读更多...
中英双语介绍大英博物馆（British Museum）

中英双语介绍大英博物馆（British Museum）

中文版 大英博物馆简介 大英博物馆&#xff08;British Museum&#xff09;位于英国伦敦市中心&#xff0c;成立于1753年&#xff0c;是世界上历史最悠久、规模最大的博物馆之一。以下是对大英博物馆的详细介绍&#xff0c;包括其地理位置、周边环境、知名度、镇馆之宝以及与…
阅读更多...
技术驱动：探索SpringBoot的大文件上传策略

技术驱动：探索SpringBoot的大文件上传策略

1.分片上传技术 为了处理大文件上传并保证性能&#xff0c;前后端可以使用分片上传&#xff08;也称为分块上传&#xff09;技术。 1.选择原因 分片上传&#xff08;也称为分块上传&#xff09;是一种处理大文件上传的技术&#xff0c;主要目的是提高上传的可靠性和效率。 网…
阅读更多...
论文导读 | 综述：大模型与推荐系统

论文导读 | 综述：大模型与推荐系统

最近&#xff0c;预训练语言模型&#xff08;PLM&#xff09;在自然语言处理领域取得了巨大成功&#xff0c;并逐渐引入推荐系统领域。本篇推文介绍了最近的两篇预训练语言模型和推荐系统结合的综述&#xff1a; [1] Pre-train, Prompt, and Recommendation: A Comprehensive …
阅读更多...
华夏女中师生深入同仁堂，感悟中医药文化之精髓

华夏女中师生深入同仁堂，感悟中医药文化之精髓

华夏女中师生深入同仁堂&#xff0c;感悟中医药文化之精髓 2024年7月4日下午&#xff0c;北京师范大学实验华夏女子中学15名学生在薛艳老师的带领下来到北京同仁堂中医医院&#xff0c;开展职业影随活动。何泽扬院长对她们的到来表示欢迎。随后&#xff0c;在“冯建春全国名老中…
阅读更多...
vue 模糊查询加个禁止属性

vue 模糊查询加个禁止属性

vue 模糊查询加个禁止属性 父组件通过属性传&#xff0c;是否禁止输入-------默认可以输入
阅读更多...
Docker：三、安装nginx与tomcat

Docker：三、安装nginx与tomcat

&#x1f341;安装常见服务 &#x1f332;安装nginx &#x1f9ca;1、搜索镜像 Ⅰ.hub docker上查询&#xff1a;https://hub.docker.com/_/nginx Ⅱ. 命令查询&#xff1a;docker search nginx &#x1f9ca;2、下载镜像 命令&#xff1a;docker pull nginx &#x1f9c…
阅读更多...
软件测试面试题总结（超全的）

软件测试面试题总结（超全的）

前面看到了一些面试题&#xff0c;总感觉会用得到&#xff0c;但是看一遍又记不住&#xff0c;所以我把面试题都整合在一起&#xff0c;都是来自各路大佬的分享&#xff0c;为了方便以后自己需要的时候刷一刷&#xff0c;不用再到处找题&#xff0c;今天把自己整理的这些面试题…
阅读更多...
【Spring Boot】统一数据返回

【Spring Boot】统一数据返回

目录 统一数据返回一. 概念二.实现统一数据返回2.1 重写responseAdvice方法2.2 重写beforeBodyWriter方法 三. 特殊类型-String的处理四. 全部代码 统一数据返回 一. 概念 其实统一数据返回是运用了AOP&#xff08;对某一类事情的集中处理&#xff09;的思维&#xff0c;简单…
阅读更多...
Python自动化，实现自动登录并爬取商品数据，实现数据可视化

Python自动化，实现自动登录并爬取商品数据，实现数据可视化

关于如何使用Python自动化登录天 猫并爬取商品数据的指南&#xff0c;我们需要明确这是一个涉及多个步骤的复杂过程&#xff0c;且需要考虑到天猫的反爬虫策略。以下是一个简化的步骤指南&#xff1a; 步骤一&#xff1a;准备工作 环境准备&#xff1a;确保你的Python环境已经…
阅读更多...
《C语言》认识数据类型和理解变量

《C语言》认识数据类型和理解变量

&#x1f339;个人主页&#x1f339;&#xff1a;喜欢草莓熊的bear &#x1f339;专栏&#x1f339;&#xff1a;C语言基础 目录 前言 一、数据类型的介绍 1.1 字符型 1.2 整形 1.3 浮点型 1.4 布尔类型 1.5 各种数据类型的长度 1.5.1 sizeof操作符 1.5.2 数据类型长度…
阅读更多...
全端面试题15（canvas）

全端面试题15（canvas）

在前端开发领域&#xff0c;<canvas> 元素和相关的 API 是面试中经常被提及的主题。下面是一些常见的关于 HTML5 Canvas 的面试问题及解答示例&#xff1a; 1. 什么是 <canvas> 元素&#xff1f; <canvas> 是 HTML5 引入的一个用于图形渲染的标签。它本身并…
阅读更多...
【Revit二次开发】创建rvt文件，但不打开Revit

【Revit二次开发】创建rvt文件，但不打开Revit

介绍 需要安装Revit&#xff0c;但不用打开Revit加载插件&#xff0c;而是运行一个控制台应用&#xff0c;就可以创建一个rvt文件&#xff08;更多读写功能都可自行添加&#xff09;。 本文内容主要参考&#xff1a;博客1&#xff0c;但对内容进行了简化&#xff0c;只保留了…
阅读更多...
PCL从理解到应用【03】KDTree 原理分析 | 案例分析 | 代码实现

PCL从理解到应用【03】KDTree 原理分析 | 案例分析 | 代码实现

前言 本文分析KDTree的原理&#xff0c;集合案例深入理解&#xff0c;同时提供源代码。 三个案例&#xff1a;K近邻搜索、半径内近邻搜索、近似最近邻搜索。方法对比&#xff0c;如下表所示&#xff1a; 特性K近邻搜索半径内近邻搜索近似最近邻搜索描述查找K个最近邻点查找指…
阅读更多...
西安邀请媒体报道，附媒体名单

西安邀请媒体报道，附媒体名单

传媒如春雨&#xff0c;润物细无声&#xff0c;大家好&#xff0c;我是51媒体网胡老师。 媒体宣传加速季&#xff0c;100万补贴享不停&#xff0c;一手媒体资源&#xff0c;全国100城线下落地执行。详情请联系胡老师。 西安市&#xff0c;作为中国古都之一&#xff0c;不仅拥有…
阅读更多...
Astro新前端框架首次体验

Astro新前端框架首次体验

Astro新前端框架首次体验 1、什么是Astro Astro是一个静态网站生成器的前端框架&#xff0c;它提供了一种新的开发方式和更好的性能体验&#xff0c;帮助开发者更快速地构建现代化的网站和应用程序。 简单来说就是&#xff1a;Astro这个是一个网站生成器&#xff0c;可以直接…
阅读更多...
DisFormer：提高视觉动态预测的准确性和泛化能力

DisFormer：提高视觉动态预测的准确性和泛化能力

最新的研究进展已经显示出目标中心的表示方法在视觉动态预测任务中可以显著提升预测精度&#xff0c;并且增加模型的可解释性。这种表示方法通过将视觉场景分解为独立的对象&#xff0c;有助于模型更好地理解和预测场景中的变化。 尽管在静态图像的解耦表示学习方面已经取得了一…
阅读更多...
基于SpringBoot的乐校园二手书交易管理系统

基于SpringBoot的乐校园二手书交易管理系统

你好呀&#xff0c;我是计算机学姐码农小野&#xff01;如果有相关需求&#xff0c;可以私信联系我。 开发语言 Java 数据库 MySQL 技术 SpringBoot框架 工具 Visual Studio、MySQL数据库开发工具 系统展示 首页 用户注册界面 二手图书界面 个人中心界面 摘要 乐校园…
阅读更多...
go Channel原理 (四)

go Channel原理 (四)

Channel 设计原理 不要通过共享内存的方式进行通信&#xff0c;而是应该通过通信的方式共享内存。 在主流编程语言中&#xff0c;多个线程传递数据的方式一般都是共享内存。 Go 可以使用共享内存加互斥锁进行通信&#xff0c;同时也提供了一种不同的并发模型&#xff0c;即通…
阅读更多...
试用笔记之-VB微信支付支付宝支付源代码

试用笔记之-VB微信支付支付宝支付源代码

首先下载VB微信支付&支付宝支付源代码 http://www.htsoft.com.cn/download/VB6WeiXin_ZhiFuBao_ZhiFu.rar
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023