RocketMQ生产者消息发送出去了,消费者一直接收不到怎么办?(Rocket MQ订阅关系一致性)
问题: 使用RocketMQ消息队列,生产者将数据发送出去了,但是生产者一致没接收到(或者是间隔好几分钟,突然接收到一条数据)怎么办?并且通过rocket web控制台查看消息的状态为NOT_ONELINE或者NOT_CONSUME,(如下图) 这种诡异现象该怎么解决?

1. 先说解决方案
这种情况99%是由于订阅关系不一致导致的,可以排查下程序看看是否有多个消费者使用了同一个group,并且订阅了不同的主题。逻辑图展示如下:

这种情况只需要将不同的消费者的group区分一下即可, 逻辑关系图变成如下这种:

到此为止,是不是惊奇的发现,问题解决了?
2. 注意事项:订阅关系一致性
看下Rocket MQ官方文档给出的说明:
定义
消费者分组是 Apache RocketMQ 系统中承载多个消费行为一致的消费者的负载均衡分组。和消费者不同,消费者分组并不是运行实体,而是一个逻辑资源。在 Apache RocketMQ 中,通过消费者分组内初始化多个消费者实现消费性能的水平扩展以及高可用容灾。

这里面只描述出了Tag的一致,事实上下面这种订阅关系也是错误的,同一个group中的两个消费者分别订阅了不同的主题, 违背了定义中的消费行为一致原则:
//Consumer c1
Consumer c1 = ConsumerBuilder.build(groupA);
c1.subscribe(topicA);
//Consumer c2Consumer
c2 = ConsumerBuilder.build(groupA);
c2.subscribe(topicB);
3. 剖析源码实现,分析原因
从GitHub下载rocketmq源码通过idea打开之后,从官方提供的example进来:

进入到DefaultMQPushConsumer构造方法中,可以发现初始化了一个DefaultMQPushConsumerImpl类:
public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,AllocateMessageQueueStrategy allocateMessageQueueStrategy) {this.consumerGroup = consumerGroup;this.namespace = namespace;this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;// 这里初始化一个默认的push类型的Consumer实现类defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);}
然后继续进入到DefaultMQPushConsumerImpl类中, 可以看见有一个成员变量MQClientInstance mQClientFactory, 在DefaultMQPushConsumerImpl类的start()(启动消费者)方法中会通过MQClientManager初始化MQClientInstance类.

接着跳转到MQClientInstance构造方法中, 会发现有这样一行代码, 初始化了一个rebalanceService. 这个rebalanceService就是RocketMQ隔一段时间进行rebalance的核心实现.

继续剖析RebalanceService类, 发现其实现了Runnable接口, 话不多说, 直接看其 run()方法中做了什么事.
呀! 原来是隔一段时间调用一次上述咱们提到的DefaultMQPushConsumerImpl类中的doRebalance()方法, 搞了半天又绕回来了. … … … … … …
直接进入到这里面, 看看rebalance的逻辑:

集群部署模式下, 会进行rebalance操作, 根据topic名称和group名称获取到所有的consumer列表.
case CLUSTERING: {Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);// 这里根据topic名称和Group进行获取到所有的consumerList<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);if (null == mqSet) {if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {this.messageQueueChanged(topic, Collections.<MessageQueue>emptySet(), Collections.<MessageQueue>emptySet());log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);}}
但是进去这行代码里面发现, topic名称仅仅用来获取Broker的网络地址, 真正获取到所有Consumer列表的是通过Group名称获取的, 看到这里相信大家基本上能够恍然大悟. 回归到上面的问题: 如果一个一个Group中的多个消费者分别订阅了不同的主题, 即: 消费行为不一致, 无论这个属于当前Group中的消费者是否订阅了这个主题, 都会参与rebalance.

画图解释一下, 假设在同一个Group下, 两个Consumer都分别订阅了Topic1和Topic2, 这种情况订阅关系一致,

假设消费者1消费Topic2的速度比较快, 经过一次rebalance之后, Consumer订阅的队列逻辑有可能成为这样的:

此时由于订阅关系的一致性, 整体系统并不会出现问题. 接下来看一种情况, 同一个消费组中的Consumer1 订阅了Topic1, Consumer2订阅了Topic2, 初始情况逻辑关系是这样:

由于进行rebalance是通过Group获取对应的消费者客户端ID, 因此rebalance之后可能出现Consumer1 指向了Topic2中的某一个队列, 同理, Consumer2指向了Topic1中的队列. 但是这与Consumer中设定的topic不一致, 因此会出现RocketMQ中消息状态为为NOT_COMSUME_YET
(个人通过对源码的简单梳理总结的文章, 如有错误欢迎指正)
相关文章:
RocketMQ生产者消息发送出去了,消费者一直接收不到怎么办?(Rocket MQ订阅关系一致性)
问题: 使用RocketMQ消息队列,生产者将数据发送出去了,但是生产者一致没接收到(或者是间隔好几分钟,突然接收到一条数据)怎么办?并且通过rocket web控制台查看消息的状态为NOT_ONELINE或者NOT_CONSUME&#…...
使用Golang开发硬件驱动
1. 介绍 Golang是一种简洁、高效的编程语言,它的强大并发性能和丰富的标准库使得它成为了开发硬件驱动的理想选择。在本文中,我们将探讨如何使用Golang开发硬件驱动程序,并提供一个实例来帮助你入门。 2. 准备工作 在开始之前,…...
设计模式(19)命令模式
一、介绍: 1、定义:命令模式(Command Pattern)是一种行为设计模式,它将请求封装为一个对象,从而使你可以使用不同的请求对客户端进行参数化。命令模式还支持请求的排队、记录日志、撤销操作等功能。 2、组…...
QModelIndex 与QStandardItem相互转换
目录 1、 QModelIndex 转换成QStandardItem 2 、QStandardItem 转换成 QModelIndex 3、示例 4、总结 1、 QModelIndex 转换成QStandardItem QStandardItem * itemQStandardItemModel::itemFromIndex(const QModelIndex & index) const 借助QStandardItemModel来完成…...
Linux - 进程地址空间
前言 首先,我们先要对 内存当中存储 各个数据之间的 结构要有一个 大概的了解: 各个区当中存储的数据使用类型不同,所以,这些数据在使用方式上是有差别的。比如下面这个例子: 在C 语言当中我们不能直接对 上述的 str…...
系统架构设计师-第16章-嵌入式系统架构设计理论与实践-软考学习笔记
嵌入式系统( Embedded System) 是为了特定应用而专门构建的计算机系统,其架构是随着嵌入式系统的逐步应用而发展形成的。嵌入式软件架构的设计与嵌入式系统的体系架构是密不可分的。因此,本常首先介绍嵌入式系统硬件相关知识(系统特征、硬件组…...
pod进阶
目录 资源限制 CPU 资源单位 内存 资源单位 实例 健康检查 探针的三种规则: Probe支持三种检查方法: 示例1:exec方式 示例2:httpGet方式 示例3:tcpSocket方式 示例4:就绪检测 扩展 资源限制 当定…...
系列四十七、Spring的事务传播行为案例演示(七)#NOT_SUPPORTED
一、演示Spring的传播行为(NOT_SUPPORTED) 1.1、StockServiceImplNOT_SUPPORTED /*** Author : 一叶浮萍归大海* Date: 2023/10/30 15:43* Description: 演示NOT_SUPPORTED的传播行为* 外部不存在事务:不开启新的事务* 外部存在…...
54.RabbitMQ快速实战以及核心概念详解
MQ MQ:MessageQueue,消息队列。这东西分两个部分来理解: 队列,是一种FIFO 先进先出的数据结构。 消息:在不同应用程序之间传递的数据。将消息以队列的形式存储起来,并且在不同的应用程序之间进行传递&am…...
Qt TreeView 设置节点不可编辑
目录 1. 创建treeview 2、节点不可编辑 3、设置logo 4、实例代码 1. 创建treeview //声明模型 QStandardItemModel *model;//创建4行,1列的模型 model new QStandardItemModel(4,1);//添加标题 model->setHeaderData(0, Qt::Horizontal, tr("Tree View…...
python django获取某个角色的某个数据和——例如:获取所有订单的应付金额总和
model关系如下: class Order(models.Model):订单product models.ForeignKey(Product, on_deletemodels.SET_NULL, blankTrue, nullTrue, verbose_name"产品")no models.CharField(max_length50, blankTrue, nullTrue, verbose_name订单编号, db_indexT…...
如何在React项目中引用less
安装less npm install less less-loader --save-dev暴露 webpack 文件 利用 npx create-react-app 搭建的 React 项目,默认隐藏 webpack 配置文件,引入 less 需要修改 webpack 配置文件,因此我们需要执行命令暴露 webpack 配置文件。 请先将…...
NUXT前端服务端渲染技术框架
服务端渲染又称SSR(Server Side Render)实在服务端完成页面的内容,而不是在客户端通过AJAX获取数据 优势:更好的SEO,由于搜索引擎爬虫抓取工具可以直接查看完全渲染的页面 Nuxt.js是一个基于Vue.js的轻量级应用框架&a…...
力扣每日一题90:子集
题目描述: 给你一个整数数组 nums ,其中可能包含重复元素,请你返回该数组所有可能的子集(幂集)。 解集 不能 包含重复的子集。返回的解集中,子集可以按 任意顺序 排列。 示例 1: 输入&#x…...
「linux基础」上传代码到github/gitee
一、在gitee创建一个仓库 1.创建仓库 2.获取仓库地址 二、克隆仓库文件到linux中 1.查看Linux中是否安装git:git --version 如果没有,在root下使用指令 yum install -y git 安装。 2.使用 git clone 仓库地址,克隆仓库文件到linux中 三、第…...
Hafnium总体考虑
安全之安全(security)博客目录导读 目录 一、安全世界构建平台 二、安全分区调度 三、平台拓扑...
C#__对Json文件的解析和序列化
Json: 存储和交换文本信息的语法。(类似XML,语法独立) 一种轻量级的数据交换格式。(更小,更快,更易解析) 语法规则: 数据在键值对里面,数据由逗号分隔开。 …...
如果一定要在C++和JAVA中选择,是C++还是java?
如果一定要在C和JAVA中选择,是C还是java? 计算机专业的同学对这个问题有疑惑的,- 定要看一下这个回答! 上来直接给出最中肯的建议: 如果你是刚刚步入大学的大一时间非常充裕的同学,猪学长强烈建议先学C/C.因为C 非常 最近很多…...
如何运行深度学习项目代码
运行项目代码是第一步哦! 配环境 使用anaconda环境; conda 环境 按照项目提示的README.md,安装指定版本的python; 当然新版python会兼容旧版,也就是你的环境下python版本比它高也不要紧; 但是更新的pyt…...
C语言 每日一题 day9
求最大值及其下标 本题要求编写程序,找出给定的n个数中的最大值及其对应的最小下标(下标从0开始)。 输入格式 : 输入在第一行中给出一个正整数n(1 < n≤10)。第二行输入n个整数,用空格分开。 输出格式 …...
简易版抽奖活动的设计技术方案
1.前言 本技术方案旨在设计一套完整且可靠的抽奖活动逻辑,确保抽奖活动能够公平、公正、公开地进行,同时满足高并发访问、数据安全存储与高效处理等需求,为用户提供流畅的抽奖体验,助力业务顺利开展。本方案将涵盖抽奖活动的整体架构设计、核心流程逻辑、关键功能实现以及…...
解锁数据库简洁之道:FastAPI与SQLModel实战指南
在构建现代Web应用程序时,与数据库的交互无疑是核心环节。虽然传统的数据库操作方式(如直接编写SQL语句与psycopg2交互)赋予了我们精细的控制权,但在面对日益复杂的业务逻辑和快速迭代的需求时,这种方式的开发效率和可…...
如何为服务器生成TLS证书
TLS(Transport Layer Security)证书是确保网络通信安全的重要手段,它通过加密技术保护传输的数据不被窃听和篡改。在服务器上配置TLS证书,可以使用户通过HTTPS协议安全地访问您的网站。本文将详细介绍如何在服务器上生成一个TLS证…...
2025 后端自学UNIAPP【项目实战:旅游项目】6、我的收藏页面
代码框架视图 1、先添加一个获取收藏景点的列表请求 【在文件my_api.js文件中添加】 // 引入公共的请求封装 import http from ./my_http.js// 登录接口(适配服务端返回 Token) export const login async (code, avatar) > {const res await http…...
sqlserver 根据指定字符 解析拼接字符串
DECLARE LotNo NVARCHAR(50)A,B,C DECLARE xml XML ( SELECT <x> REPLACE(LotNo, ,, </x><x>) </x> ) DECLARE ErrorCode NVARCHAR(50) -- 提取 XML 中的值 SELECT value x.value(., VARCHAR(MAX))…...
BCS 2025|百度副总裁陈洋:智能体在安全领域的应用实践
6月5日,2025全球数字经济大会数字安全主论坛暨北京网络安全大会在国家会议中心隆重开幕。百度副总裁陈洋受邀出席,并作《智能体在安全领域的应用实践》主题演讲,分享了在智能体在安全领域的突破性实践。他指出,百度通过将安全能力…...
MySQL中【正则表达式】用法
MySQL 中正则表达式通过 REGEXP 或 RLIKE 操作符实现(两者等价),用于在 WHERE 子句中进行复杂的字符串模式匹配。以下是核心用法和示例: 一、基础语法 SELECT column_name FROM table_name WHERE column_name REGEXP pattern; …...
mysql已经安装,但是通过rpm -q 没有找mysql相关的已安装包
文章目录 现象:mysql已经安装,但是通过rpm -q 没有找mysql相关的已安装包遇到 rpm 命令找不到已经安装的 MySQL 包时,可能是因为以下几个原因:1.MySQL 不是通过 RPM 包安装的2.RPM 数据库损坏3.使用了不同的包名或路径4.使用其他包…...
均衡后的SNRSINR
本文主要摘自参考文献中的前两篇,相关文献中经常会出现MIMO检测后的SINR不过一直没有找到相关数学推到过程,其中文献[1]中给出了相关原理在此仅做记录。 1. 系统模型 复信道模型 n t n_t nt 根发送天线, n r n_r nr 根接收天线的 MIMO 系…...
Python基于历史模拟方法实现投资组合风险管理的VaR与ES模型项目实战
说明:这是一个机器学习实战项目(附带数据代码文档),如需数据代码文档可以直接到文章最后关注获取。 1.项目背景 在金融市场日益复杂和波动加剧的背景下,风险管理成为金融机构和个人投资者关注的核心议题之一。VaR&…...
