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

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

目录

前言

在踏入投递消息 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 结构体中可以设置的属性如下：

字段名	默认值	必要性	说明
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？

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

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() {
  @Override
  public void onSuccess(SendResult sendResult) {
    System.out.println("message send success...");
  }

  @Override
  public 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 中的配置参数：orderMessageEnable、returnOrderTopicConfigToBroker 必须是 true，如果上述任意一个条件不满足，则是保证可用性而不是严格顺序

总结

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

学习指南针：

Rocket 官方文档

RocketMQ GitHub

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

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

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

推荐专栏：Spring、MySQL，订阅一波不再迷路

大家的「关注?? + 点赞👍 + 收藏?」就是我创作的最大动力！谢谢大家的支持，我们下文见！