🔭 嗨，您好 👋 我是 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 消息构成的结构如下：
- Topic：表示要发送消息的主题
- body：表示消息的存储内容
- Properties：表示消息属性
- transactionId：在事务消息中使用
Keys、Tag：无论是 RocketMQ Tag 过滤还是延迟消息等都会利用 Properties 消息属性机制，这些特殊信息都使用了系统保留的属性 Key，设置自定义属性时要避免和系统属性 Key 相冲突
服务端会根据 Keys 创建哈希索引，设置以后，可以在 Console 系统通过 Topic、Keys 来查询消息，由于是哈希索引，尽可能保证 Key 唯一，例如：订单号、会员 Id 等
RocketMQ 系统保留的属性 Key 集合，在使用消息属性机制时尽量避免：org.apache.rocketmq.common.message.MessageConst
Properties
在 Message 结构体中可以设置的属性如下：
| 字段名 | 默认值 | 必要性 | 说明 |
|---|---|---|---|
| Topic | null | 必填 | 消息所属 Topic 名称 |
| Body | null | 必填 | 消息体 |
| Tags | null | 选填 | 消息标签：方便服务器过滤消息使用 消息属性：TAGS |
| Keys | null | 选填 | 代表这条消息的业务关键词 消息属性：KEYS |
| Flag | 0 | 选填 | 由应用程序设置，RocketMQ 不作干预 |
| DelayTimeLevel | 0 | 选填 | 消息延迟级别，0 表示不延时， 大于 0 会延迟特定的时间才会被消息 消息属性：DELAY |
| WaitStoreMsgOK | true | 选填 | 表示消息是否在服务器落盘后才返回 SEND_OK 消息属性：WAIT |
Topic、Tag
Topic、Tag 都是用来在业务上分类的标识，区别在于 Topic 是一级分类，Tag 理解为是二级分类，使用 Tag 可以实现对 Topic 中的消息进行过滤
Topic：消息主题，通过 Topic 对不同的业务消息进行分类
Tag：消息标签，用来进一步区分某个 Topic 下的消息分类，消息从生产者发出时就带上的属性
Tag 属性可以为 Topic 做另外一层业务上再细粒化地区分子业务
什么时候该用 Topic，什么该用 Tag？
一般可以从以下几方面进行判别：
-
消息类型是否一致：普通消息、事务消息、定时(延时)消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分
除了普通消息以外，其余的消息类型应该使用对应的后缀拼接，如：事务 _transaction、延迟 _delay、顺序 _order
-
业务是否相关联：没有直接关联的消息，如淘宝交易消息与京东物流消息使用不同的 Topic 区分；若同样的是京东物流消息，电器类订单、男装类订单、鞋类订单的消息可以使用 Tag 进行划分
-
消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分
-
消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 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 通过生产者和服务端的协议保障 单个生产者以串行的方式 发送消息，并按序存储和持久化，如需保证消息生产的有序性，则必须满足以下条件：
- 单一生产者：消息的生产的顺序仅支持单一生产者，不同生产者分布在不同的系统，即使设置为相同的分区键-arg，不同生产者之间产生的消息也无法判定其顺序
- 串行发送：生产者客户端支持多线程安全访问，但如果生产者使用了多线程并行发送，则不同线程间产生的消息将无法判定其先后顺序
满足以上条件的生产者，将顺序消息发送至服务端以后，会保证设置了同一分区键的消息，按照发送的顺序存储在同一个队列中
在 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 中的配置参数：orderMessageEnable、returnOrderTopicConfigToBroker 必须是 true，如果上述任意一个条件不满足，则是保证可用性而不是严格顺序
总结
该篇博文主要介绍 Message 结构体有哪些以及它里面属性的作用，同步与异步发送之间的区别参数：SendCallback 回调接口，指定 Queue 投递消息：MessageQueue，生产者侧确保消息能够有序地投递：MessageQueueSelector，以及在 RocketMQ 如何保证全局有序、局部有序：生产者确保单 Queue 全局有序、生产者确保多 Queue 局部有序，希望该篇博文你能够喜欢，感谢三连支持❤️
学习指南针：
Rocket 官方文档
RocketMQ GitHub
🌟🌟🌟愿你我都能够在寒冬中相互取暖，互相成长，只有不断积累、沉淀自己，后面有机会自然能破冰而行！
博文放在 RocketMQ 专栏里，欢迎订阅，会持续更新！
如果觉得博文不错，关注我 vnjohn，后续会有更多实战、源码、架构干货分享！
推荐专栏：Spring、MySQL，订阅一波不再迷路
大家的「关注❤️ + 点赞👍 + 收藏⭐」就是我创作的最大动力！谢谢大家的支持，我们下文见！
