RocketMQ 投递消息方式以及消息体结构分析:Message、MessageQueueSelector

本文主要是介绍RocketMQ 投递消息方式以及消息体结构分析:Message、MessageQueueSelector,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

在这里插入图片描述

🔭 嗨,您好 👋 我是 vnjohn,在互联网企业担任 Java 开发,CSDN 优质创作者
📖 推荐专栏:Spring、MySQL、Nacos、Java,后续其他专栏会持续优化更新迭代
🌲文章所在专栏:RocketMQ
🤔 我当前正在学习微服务领域、云原生领域、消息中间件等架构、原理知识
💬 向我询问任何您想要的东西,ID:vnjohn
🔥觉得博主文章写的还 OK,能够帮助到您的,感谢三连支持博客🙏
😄 代词: vnjohn
⚡ 有趣的事实:音乐、跑步、电影、游戏

目录

  • 前言
  • Message
    • Properties
    • Topic、Tag
    • Keys
    • Queue
  • SendCallback
  • MessageQueue
  • MessageQueueSelector
  • FAQ
  • 总结

前言

在踏入投递消息 send 方法源码解读开始前,要先搞清楚消息的结构以及消息的不同的方式实现,以便于在阅读底层源码时不至于要重头过来再梳理各自的类、属性结构起到了什么作用,在这里再啰嗦啰嗦,让 RocketMQ 源码能够更加完整.

RocketMQ 专栏篇:

从零开始:手把手搭建 RocketMQ 单节点、集群节点实例

保护数据完整性:探索 RocketMQ 分布式事务消息的力量

RocketMQ 分布式事务消息实战指南:确保数据一致性的关键设计

RocketMQ 生产者源码分析:DefaultMQProducer、DefaultMQProducerImpl

RocketMQ MQClientInstance、生产者实例启动源码分析

Message

RocketMQ 消息构成的结构如下:

  1. Topic:表示要发送消息的主题
  2. body:表示消息的存储内容
  3. Properties:表示消息属性
  4. transactionId:在事务消息中使用

Keys、Tag:无论是 RocketMQ Tag 过滤还是延迟消息等都会利用 Properties 消息属性机制,这些特殊信息都使用了系统保留的属性 Key,设置自定义属性时要避免和系统属性 Key 相冲突

服务端会根据 Keys 创建哈希索引,设置以后,可以在 Console 系统通过 Topic、Keys 来查询消息,由于是哈希索引,尽可能保证 Key 唯一,例如:订单号、会员 Id 等

RocketMQ 系统保留的属性 Key 集合,在使用消息属性机制时尽量避免:org.apache.rocketmq.common.message.MessageConst

Properties

在 Message 结构体中可以设置的属性如下:

字段名默认值必要性说明
Topicnull必填消息所属 Topic 名称
Bodynull必填消息体
Tagsnull选填消息标签:方便服务器过滤消息使用
消息属性:TAGS
Keysnull选填代表这条消息的业务关键词
消息属性:KEYS
Flag0选填由应用程序设置,RocketMQ 不作干预
DelayTimeLevel0选填消息延迟级别,0 表示不延时,
大于 0 会延迟特定的时间才会被消息
消息属性:DELAY
WaitStoreMsgOKtrue选填表示消息是否在服务器落盘后才返回 SEND_OK
消息属性:WAIT

Topic、Tag

Topic、Tag 都是用来在业务上分类的标识,区别在于 Topic 是一级分类,Tag 理解为是二级分类,使用 Tag 可以实现对 Topic 中的消息进行过滤

Topic:消息主题,通过 Topic 对不同的业务消息进行分类

Tag:消息标签,用来进一步区分某个 Topic 下的消息分类,消息从生产者发出时就带上的属性

Tag 属性可以为 Topic 做另外一层业务上再细粒化地区分子业务

在这里插入图片描述

什么时候该用 Topic,什么该用 Tag?

一般可以从以下几方面进行判别:

  1. 消息类型是否一致:普通消息、事务消息、定时(延时)消息、顺序消息,不同的消息类型使用不同的 Topic,无法通过 Tag 进行区分

    除了普通消息以外,其余的消息类型应该使用对应的后缀拼接,如:事务 _transaction、延迟 _delay、顺序 _order

  2. 业务是否相关联:没有直接关联的消息,如淘宝交易消息与京东物流消息使用不同的 Topic 区分;若同样的是京东物流消息,电器类订单、男装类订单、鞋类订单的消息可以使用 Tag 进行划分

  3. 消息优先级是否一致:如同样是物流消息,盒马必须小时内送达,天猫超市 24 小时内送达,淘宝物流则相对会慢一些,不同优先级的消息用不同的 Topic 进行区分

  4. 消息量级是否相当:有些业务消息虽然量小但是实时性要求高,如果跟某些万亿量级的消息使用同一个 Topic,则有可能会因为过长的等待时间而“饿死”,此时需要将不同量级的消息进行拆分,使用不同的 Topic

总的来说,针对消息分类,可以选择创建多个 Topic,或者在同一个 Topic 下创建多个 Tag;通常情况下,不同的 Topic 之间消息没有必然的联系,而 Tag 则用来区分同一个 Topic 下相互关联的消息,例如:全集、子集的关系,流程先后的关系.

Keys

RocketMQ 每个消息可以在业务层面上设置唯一标识码 Keys 字段,方便将来定位消息丢失的问题,Broker 会为每个消息创建哈希索引,应用可以通过 Topic、Key 来查询这条消息的内容,以及消息被谁消费,由于是哈希索引,请务必保证 Key 尽可能唯一,这样可以避免潜在的哈希冲突

// 订单Id
String orderId = "2024016163739";
message.setKeys(orderId);

Queue

为了支持高并发以及水平扩展,需要对 Topic 进行分区「类比于 Kafka Partition 机制」在 RocketMQ 中被称之为队列,一个 Topic 下可能有多个队列,并且可能分布在不同的 Broker 上

在这里插入图片描述

一般来说一条消息,若没有重复发送(比如:因为客户端没有响应而进行重试)则只会存在 Topic 其中的一个队列,消息在队列中按照先进先出的原则存储,每条消息会有自己的位点,每个队列会统计当前消息的总条数,这个称为最大位点:MaxOffset;队列的起始位置对应的位置叫做起始位点 MinOffset,队列可以提升消息发送和消费的并发度.

在这里谈到了 Topic、Queue 之间的关联,也就有必要说说 RocketMQ 中的 AKF 原则了

在这里插入图片描述

  • X 轴:处理的服务节点的单点故障问题,支持横向扩展、全量镜像
  • Y 轴:在 RocketMQ 服务节点基础上根据业务来划分出不同的 Topic
  • Z 轴:基于 Topic 分配出不同的 Queue 分区,每个 Queue 可能分散到不同的 Broker 服务节点上

SendCallback

当 send 方法 SendCallback 参数不为空,说明它是属于异步发送的方式,异步发送:发送方发出一条消息以后,不等待服务端返回响应,会接着发送下一条消息

SendCallback 是基于异步发送时需要的指定的回调接口,它提供了两个方法:

public interface SendCallback {void onSuccess(final SendResult sendResult);void onException(final Throwable e);
}

消息发送方在发送了一条消息后,不需要等待服务端响应即可发送第二条消息,发送方通过回调接口接收服务端响应,并处理响应结果。异步发送一般用于链路耗时较长,对响应时间较为敏感的业务场景。例如:视频上传后通知启动转码服务,转码完成后通知推送转码结果等

同步发送示例代码:

DefaultMQProducer mqProducer = new DefaultMQProducer();
mqProducer.setProducerGroup("vnjohn");
mqProducer.setNamesrvAddr("172.16.249.10:9876");
mqProducer.start();
Message message = new Message();
message.setTopic("vnjohn");
message.setWaitStoreMsgOK(Boolean.FALSE);
message.setKeys("2024016163739");
message.setBody("Hello RocketMQ".getBytes(StandardCharsets.UTF_8));
message.setTags("blog");
mqProducer.send(message);

异步发送示例代码:

mqProducer.send(message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("message send success...");}@Overridepublic void onException(Throwable e) {System.out.println("message send exception...");}
});

异步发送、同步发送代码唯一区别在于调用 send 接口的参数不同,异步发送不会等待发送返回,取而代之的是 send 方法需要传入 SendCallback 的实现,SendCallback 接口主要有 onSuccess、onException 两个方法,表示消息发送成功和消息发送失败

MessageQueue

当 send 方法时,指定了 MessageQueue 参数,说明要将消息指定投递给 Topic 其中的一个队列:Queue

// MessageQueue:指定 Topic 主题、Broker、Topic-Queue
private String topic;
private String brokerName;
private int queueId;

比如我要将消息投递给 「Topic:vnjohn」 下的 「Queue:0」所属的 「brokerName:broker-0」

mqProducer.send(message, new MessageQueue("vnjohn", "broker-0", 0));

MessageQueueSelector

在生产者投递消息时,通过该类可以指定将需要保证顺序消费的消息「通过 args 来指定」放入到同一个队列中,确保在消费时可以通过消费这一个队列保证消费的顺序性

生产顺序性:

RocketMQ 通过生产者和服务端的协议保障 单个生产者以串行的方式 发送消息,并按序存储和持久化,如需保证消息生产的有序性,则必须满足以下条件:

  1. 单一生产者:消息的生产的顺序仅支持单一生产者,不同生产者分布在不同的系统,即使设置为相同的分区键-arg,不同生产者之间产生的消息也无法判定其顺序
  2. 串行发送:生产者客户端支持多线程安全访问,但如果生产者使用了多线程并行发送,则不同线程间产生的消息将无法判定其先后顺序

满足以上条件的生产者,将顺序消息发送至服务端以后,会保证设置了同一分区键的消息,按照发送的顺序存储在同一个队列中

在 RocketMQ 中,Topic 下只存在一个 Queue 时,生产者可以保证 全局有序性,否则生产者就只能保证 局部有序性

对于一个指定的 Topic,消息严格按照先进先出(FIFO)的原则进行消息发布和消费,即先发布的消息先消费,后发布的消息后消费。在 RocketMQ 中支持分区顺序消息,如下图所示:我们可以按照某一个标准对消息进行分区(比如图中的 ShardingKey),同一个 ShardingKey 消息会被分配到同一个队列中,并按照顺序被消费

在这里插入图片描述

顺序消息的应用场景非常广泛,在有序事件处理、撮合交易、数据实时增量同步等场景下,异构系统之间需要维持强一致的状态同步,上游的事件变更需要按照顺序传递到下游进行处理

例如:创建订单的场景,需要保证同一个订单的生成、付款和发货,这三个操作被顺序执行;如果是普通消息,订单-A 的消息可能会被轮询发送到不同的队列中,不同队列的消息将无法保证顺序,而顺序消息发送时将 ShardingKey 相同(同一个订单号)的消息路由到同一个逻辑队列中

提到的 SharingKey 指的就是在往服务端有序发送消息时,必须指定的一个参数:Object arg,假设它就是订单 id

在这里插入图片描述

MessageQueueSelector 是一个接口,它的定义如下:

public interface MessageQueueSelector {MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

其中 mqs 是可以发送的队列,msg 是消息,arg 是 send 方法时传递的 Object 对象,返回的是该消息需要发送到的队列

通过 MessageQueueSelector 顺序发送消息的示例代码如下:

MessageQueueSelector customMessageQueueSelector = (mqs, msg, arg) -> {int shardingKey = arg.hashCode();int messageQueueIndex = mqs.size() % shardingKey;return mqs.get(messageQueueIndex);
};
mqProducer.send(message, customMessageQueueSelector, "20240107015444");

生产环境中建议选择最细粒度的分区键进行拆分,例如:将订单 ID、用户 ID 作为分区键关键字,可实现同一个终端用户的消息按照顺序处理,不同用户的消息无须保证顺序.

FAQ

若一个 Broker 掉线,队列总数发送变化会如何?

若发送变化,那么同一个 ShardingKey 的消息就会发送到不同的队列上,造成乱序;若不发生变化,那消息将会发送到掉线 Broker 队列上,必然是失败的;因此 RocketMQ 提供了两种模式:严格顺序、顺序可用性,如果要保证严格顺序而不是可用性,创建 Topic 时要指定 -o (-order)参数,表示顺序消息主题

$ sh bin/mqadmin updateTopic -c DefaultCluster -t TopicTest -o true -n 127.0.0.1:9876
create topic to 127.0.0.1:10911 success.
TopicConfig [topicName=TopicTest, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=true, attributes=null]

其次就是要保证 nameserver 中的配置参数:orderMessageEnablereturnOrderTopicConfigToBroker 必须是 true,如果上述任意一个条件不满足,则是保证可用性而不是严格顺序

总结

该篇博文主要介绍 Message 结构体有哪些以及它里面属性的作用,同步与异步发送之间的区别参数:SendCallback 回调接口,指定 Queue 投递消息:MessageQueue,生产者侧确保消息能够有序地投递:MessageQueueSelector,以及在 RocketMQ 如何保证全局有序、局部有序:生产者确保单 Queue 全局有序、生产者确保多 Queue 局部有序,希望该篇博文你能够喜欢,感谢三连支持❤️

学习指南针:

Rocket 官方文档

RocketMQ GitHub

🌟🌟🌟愿你我都能够在寒冬中相互取暖,互相成长,只有不断积累、沉淀自己,后面有机会自然能破冰而行!

博文放在 RocketMQ 专栏里,欢迎订阅,会持续更新!

如果觉得博文不错,关注我 vnjohn,后续会有更多实战、源码、架构干货分享!

推荐专栏:Spring、MySQL,订阅一波不再迷路

大家的「关注❤️ + 点赞👍 + 收藏⭐」就是我创作的最大动力!谢谢大家的支持,我们下文见!

这篇关于RocketMQ 投递消息方式以及消息体结构分析:Message、MessageQueueSelector的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/579240

相关文章

字节面试 | 如何测试RocketMQ、RocketMQ?

字节面试:RocketMQ是怎么测试的呢? 答: 首先保证消息的消费正确、设计逆向用例,在验证消息内容为空等情况时的消费正确性; 推送大批量MQ,通过Admin控制台查看MQ消费的情况,是否出现消费假死、TPS是否正常等等问题。(上述都是临场发挥,但是RocketMQ真正的测试点,还真的需要探讨) 01 先了解RocketMQ 作为测试也是要简单了解RocketMQ。简单来说,就是一个分

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题,本文将一起从代码层去分析为什么没有建立索引? 开源ERP项目地址:https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置

内核启动时减少log的方式

内核引导选项 内核引导选项大体上可以分为两类:一类与设备无关、另一类与设备有关。与设备有关的引导选项多如牛毛,需要你自己阅读内核中的相应驱动程序源码以获取其能够接受的引导选项。比如,如果你想知道可以向 AHA1542 SCSI 驱动程序传递哪些引导选项,那么就查看 drivers/scsi/aha1542.c 文件,一般在前面 100 行注释里就可以找到所接受的引导选项说明。大多数选项是通过"_

usaco 1.3 Mixing Milk (结构体排序 qsort) and hdu 2020(sort)

到了这题学会了结构体排序 于是回去修改了 1.2 milking cows 的算法~ 结构体排序核心: 1.结构体定义 struct Milk{int price;int milks;}milk[5000]; 2.自定义的比较函数,若返回值为正,qsort 函数判定a>b ;为负,a<b;为0,a==b; int milkcmp(const void *va,c

用命令行的方式启动.netcore webapi

用命令行的方式启动.netcore web项目 进入指定的项目文件夹,比如我发布后的代码放在下面文件夹中 在此地址栏中输入“cmd”,打开命令提示符,进入到发布代码目录 命令行启动.netcore项目的命令为:  dotnet 项目启动文件.dll --urls="http://*:对外端口" --ip="本机ip" --port=项目内部端口 例: dotnet Imagine.M

SWAP作物生长模型安装教程、数据制备、敏感性分析、气候变化影响、R模型敏感性分析与贝叶斯优化、Fortran源代码分析、气候数据降尺度与变化影响分析

查看原文>>>全流程SWAP农业模型数据制备、敏感性分析及气候变化影响实践技术应用 SWAP模型是由荷兰瓦赫宁根大学开发的先进农作物模型,它综合考虑了土壤-水分-大气以及植被间的相互作用;是一种描述作物生长过程的一种机理性作物生长模型。它不但运用Richard方程,使其能够精确的模拟土壤中水分的运动,而且耦合了WOFOST作物模型使作物的生长描述更为科学。 本文让更多的科研人员和农业工作者

MOLE 2.5 分析分子通道和孔隙

软件介绍 生物大分子通道和孔隙在生物学中发挥着重要作用,例如在分子识别和酶底物特异性方面。 我们介绍了一种名为 MOLE 2.5 的高级软件工具,该工具旨在分析分子通道和孔隙。 与其他可用软件工具的基准测试表明,MOLE 2.5 相比更快、更强大、功能更丰富。作为一项新功能,MOLE 2.5 可以估算已识别通道的物理化学性质。 软件下载 https://pan.quark.cn/s/57

自定义类型:结构体(续)

目录 一. 结构体的内存对齐 1.1 为什么存在内存对齐? 1.2 修改默认对齐数 二. 结构体传参 三. 结构体实现位段 一. 结构体的内存对齐 在前面的文章里我们已经讲过一部分的内存对齐的知识,并举出了两个例子,我们再举出两个例子继续说明: struct S3{double a;int b;char c;};int mian(){printf("%zd\n",s

衡石分析平台使用手册-单机安装及启动

单机安装及启动​ 本文讲述如何在单机环境下进行 HENGSHI SENSE 安装的操作过程。 在安装前请确认网络环境,如果是隔离环境,无法连接互联网时,请先按照 离线环境安装依赖的指导进行依赖包的安装,然后按照本文的指导继续操作。如果网络环境可以连接互联网,请直接按照本文的指导进行安装。 准备工作​ 请参考安装环境文档准备安装环境。 配置用户与安装目录。 在操作前请检查您是否有 sud

线性因子模型 - 独立分量分析(ICA)篇

序言 线性因子模型是数据分析与机器学习中的一类重要模型,它们通过引入潜变量( latent variables \text{latent variables} latent variables)来更好地表征数据。其中,独立分量分析( ICA \text{ICA} ICA)作为线性因子模型的一种,以其独特的视角和广泛的应用领域而备受关注。 ICA \text{ICA} ICA旨在将观察到的复杂信号