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个整数,用空格分开。 输出格式 …...
利用ngx_stream_return_module构建简易 TCP/UDP 响应网关
一、模块概述 ngx_stream_return_module 提供了一个极简的指令: return <value>;在收到客户端连接后,立即将 <value> 写回并关闭连接。<value> 支持内嵌文本和内置变量(如 $time_iso8601、$remote_addr 等)&a…...
Java 语言特性(面试系列1)
一、面向对象编程 1. 封装(Encapsulation) 定义:将数据(属性)和操作数据的方法绑定在一起,通过访问控制符(private、protected、public)隐藏内部实现细节。示例: public …...
《从零掌握MIPI CSI-2: 协议精解与FPGA摄像头开发实战》-- CSI-2 协议详细解析 (一)
CSI-2 协议详细解析 (一) 1. CSI-2层定义(CSI-2 Layer Definitions) 分层结构 :CSI-2协议分为6层: 物理层(PHY Layer) : 定义电气特性、时钟机制和传输介质(导线&#…...
pam_env.so模块配置解析
在PAM(Pluggable Authentication Modules)配置中, /etc/pam.d/su 文件相关配置含义如下: 配置解析 auth required pam_env.so1. 字段分解 字段值说明模块类型auth认证类模块,负责验证用户身份&am…...
【算法训练营Day07】字符串part1
文章目录 反转字符串反转字符串II替换数字 反转字符串 题目链接:344. 反转字符串 双指针法,两个指针的元素直接调转即可 class Solution {public void reverseString(char[] s) {int head 0;int end s.length - 1;while(head < end) {char temp …...
【C语言练习】080. 使用C语言实现简单的数据库操作
080. 使用C语言实现简单的数据库操作 080. 使用C语言实现简单的数据库操作使用原生APIODBC接口第三方库ORM框架文件模拟1. 安装SQLite2. 示例代码:使用SQLite创建数据库、表和插入数据3. 编译和运行4. 示例运行输出:5. 注意事项6. 总结080. 使用C语言实现简单的数据库操作 在…...
如何在最短时间内提升打ctf(web)的水平?
刚刚刷完2遍 bugku 的 web 题,前来答题。 每个人对刷题理解是不同,有的人是看了writeup就等于刷了,有的人是收藏了writeup就等于刷了,有的人是跟着writeup做了一遍就等于刷了,还有的人是独立思考做了一遍就等于刷了。…...
Android 之 kotlin 语言学习笔记三(Kotlin-Java 互操作)
参考官方文档:https://developer.android.google.cn/kotlin/interop?hlzh-cn 一、Java(供 Kotlin 使用) 1、不得使用硬关键字 不要使用 Kotlin 的任何硬关键字作为方法的名称 或字段。允许使用 Kotlin 的软关键字、修饰符关键字和特殊标识…...
让回归模型不再被异常值“带跑偏“,MSE和Cauchy损失函数在噪声数据环境下的实战对比
在机器学习的回归分析中,损失函数的选择对模型性能具有决定性影响。均方误差(MSE)作为经典的损失函数,在处理干净数据时表现优异,但在面对包含异常值的噪声数据时,其对大误差的二次惩罚机制往往导致模型参数…...
Docker 本地安装 mysql 数据库
Docker: Accelerated Container Application Development 下载对应操作系统版本的 docker ;并安装。 基础操作不再赘述。 打开 macOS 终端,开始 docker 安装mysql之旅 第一步 docker search mysql 》〉docker search mysql NAME DE…...
