深入理解Apache RocketMQ 中Message 消息的核心概念

深入理解一下Apache RocketMQ中Message(消息)这个核心概念,这份文档详细阐述了消息的定义、在模型中的位置、内部属性、约束和使用建议,感兴趣的朋友跟随小编一起学习吧

好的，我们来深入理解一下 Apache RocketMQ 中 Message (消息) 这个核心概念。这份文档详细阐述了消息的定义、在模型中的位置、内部属性、约束和使用建议。

推荐阅读：RocketMQ 消息Message的结构和使用方式详解

你可以将 Message 看作是 RocketMQ 系统中数据传输和处理的最小原子单位。它承载了业务数据，并附带了丰富的元信息，是生产者、Broker 和消费者之间通信的载体。

1.Message的本质定义

• 最小传输单元 (Smallest Unit of Data Transmission)：
  • Message 是 RocketMQ 中数据传输的基本单元。生产者将业务数据（负载）和扩展属性封装成 Message，发送给 Broker；Broker 再根据订阅关系将 Message 传递给消费者。
• 核心特性 (Characteristics)：
  • 不可变性 (Immutability)：
    • 消息一旦生成，其内容（特别是系统属性和负载）在传输和存储过程中不会改变。它被视为一个“已发生的事件”。
    • 消费者获取到的消息是只读的 (read-only)。在 5.x 版本中，这是强约束；在 3.x/4.x 版本中虽无强约束，但最佳实践也建议不要修改。
    • 最佳实践：如果需要基于收到的消息发送新消息，应该创建一个新消息（例如 MessageBuilder.buildFrom(m)），而不是直接修改原消息。
  • 持久性 (Persistence)：
    • 默认情况下，RocketMQ 会将接收到的消息持久化存储在 Broker 的存储文件中。这是保证消息不丢失、支持消息追溯和系统故障恢复的基础。

2.Message在模型中的位置

• 生命周期流程：
  1. 生产 (Produced)：由生产者 (Producer) 创建并初始化。
  2. 发送 (Sent)：生产者将消息发送到 Apache RocketMQ Broker。
  3. 存储 (Stored)：Broker 接收到消息后，将其按接收顺序存储在特定 Topic 的某个 Queue 中。
  4. 消费 (Consumed)：消费者 (Consumer) 根据订阅关系，从 Broker 的相应 Queue 中拉取 (pull) 消息进行消费。

3.Message的核心内部属性

这些属性分为系统保留属性 (System retention attributes) 和可选属性 (Optional attributes)，以及负载 (Load)。

系统保留属性 (由系统或生产者设置)

• Topic 名称 (Topic name)：
  • 作用：标识该消息属于哪个逻辑主题。在集群内必须唯一。
  • 来源：由生产者 SDK 设置。

• 消息类型 (Message type)：

  • 作用：定义消息的语义和处理方式。RocketMQ 支持多种类型：
    • Normal：普通消息，无特殊语义。
    • FIFO：顺序消息，保证同一消息组 (Message Group) 内的消息按发送顺序被消费。实现依赖于 Queue 的有序性。
    • Delay：延迟消息，可以指定延迟时间（最大40天），延迟时间到后才对消费者可见。
    • Transaction：事务消息，用于实现分布式事务，确保本地数据库操作和消息发送的最终一致性。

• 消息队列 (Message queue)：

  • 作用：指明该消息最终被存储在哪个具体的 Queue 中（属于哪个 Topic 的哪个 Queue）。
  • 来源：由 Broker 在消息到达后，根据路由策略（如轮询、哈希等）指定并填充。

• 消息 Offset (Message offset)：

  • 作用：标识该消息在其所属 Queue 内部的物理存储位置（偏移量）。是 Broker 内部管理消息顺序和消费者消费进度的关键。
  • 来源：由 Broker 指定并填充。从 0 开始递增。

• 消息 ID (Message ID)：

  • 作用：消息的全局唯一标识符。在集群内绝对唯一，用于消息追踪、排查问题。
  • 来源：由生产者客户端自动生成（通常是 32 位的数字和大写字母组成的字符串）。

可选属性 (由生产者设置)

• (可选) 消息 Keys (Message keys)：
  • 作用：为消息设置一个或多个索引键。主要用于消息查询（通过 Message ID 或 Message Key 在控制台或通过 API 查找特定消息）和去重（结合业务逻辑）。
  • 来源：由生产者客户端定义。
• (可选) 消息 Tag (Message tag)：
  • 作用：消息的标签，用于消费者进行消息过滤。消费者可以订阅特定的 Tag，从而只接收带有该 Tag 的消息，实现简单的消息分类。
  • 来源：由生产者客户端定义。
  • 约束：每个消息只能设置一个 Tag。
• (可选) 定时时间 (Scheduled time)：
  • 作用：配合 Delay 消息类型使用，指定消息延迟的具体时间戳（毫秒级），而不是延迟时长。
  • 来源：由消息生产者定义。
  • 约束：最大延迟时间 40 天。

时间戳属性

• 消息发送时间 (Message sending time)：
  • 作用：记录消息在生产者客户端本地被发送出去的时间戳（毫秒级）。
  • 来源：由生产者客户端填充。
  • 注意：这是客户端时间，可能与 Broker 时间有偏差。
• 消息存储时间 (Message store timestamp)：
  • 作用：记录消息被Broker 成功写入存储（落盘）的时间戳（毫秒级）。对于延迟消息和事务消息，消费者感知到的“有效时间”通常基于此时间。
  • 来源：由Broker 填充。
  • 注意：这是 Broker 时间，是消息在服务端的“出生”时间。

重试与自定义

• 重试次数 (Retry times)：
  • 作用：记录该消息被 Broker 重新投递给消费者的次数。每次消费失败触发重试，次数加一。第一次消费时为 0。
  • 来源：由 Broker 标记。消费者可以获取此信息以进行幂等处理或特殊逻辑。
• 自定义属性 (Custom attributes)：
  • 作用：生产者可以添加任意的 Key-Value (字符串类型) 对作为扩展信息，供业务逻辑使用。
  • 来源：由生产者根据需要指定。
• 消息负载 (Message load)：
  • 作用：消息的实际业务数据内容，即有效载荷 (Payload)。
  • 来源：由生产者序列化成二进制字节流后设置。
  • 约束：大小不能超过系统限制。

4.Message的行为约束

• 大小限制 (Size Limit)：
  • 核心约束：单条消息的大小不能超过上限，否则发送会失败。
  • 默认限制：4 MB。这是非常重要的参数，直接影响网络传输、存储和处理性能。

5.Message的使用建议与最佳实践

• 避免单条消息过大 (Overloaded transmission)：
  • 原因：RocketMQ 是事件驱动的中间件。过大的消息会：
    • 加重网络传输负担，增加延迟。
    • 影响错误重试：重试大消息成本高。
    • 影响流控 (Throttling)：流控粒度可能不够精细。
  • 建议：
    1. 严格控制单条消息的数据量，使其尽可能小。
    2. 如果业务上必须传输大量数据，强烈建议：
       • 拆分消息：将大数据按固定大小拆分成多条小消息。
       • 使用外部存储：将实际数据（如文件、图片）存放到对象存储（如 OSS）、文件系统或数据库中，然后在消息的 load 或 custom attributes 中只传递数据的访问链接 (URL) 或 ID。
• 遵守消息不可变性原则 (Immutability)：
  • 正确做法：收到消息后，如果需要转发或基于它生成新消息，使用 MessageBuilder.buildFrom(m) 这样的方法创建一个新消息实例，然后修改新实例的属性（如 Topic, Tag, Load 等）再发送。
  • 错误做法：直接调用 m.update() 修改收到的消息 m 的内容，然后发送。这违反了不可变性原则，可能导致不可预知的行为或在 5.x 版本中被拒绝。

总结与核心理解

1. Message 是原子单元：它封装了业务数据和元信息，是 RocketMQ 传输的最小单位。
2. Message 是不可变的：内容一旦产生，在传递过程中不应被修改。最佳实践是“读取-创建-发送”新消息。
3. Message 是持久化的：默认落盘存储，保证可靠性。
4. Message 拥有丰富的属性：
   • Topic/Queue/Offset 定义了其在系统中的位置和顺序。
   • Message ID 提供全局唯一标识。
   • Keys/Tag 支持查询和过滤。
   • Message Type 定义了语义（普通、顺序、延迟、事务）。
   • Sending Time/Store Timestamp 记录了关键时间点。
   • Retry Times 协助处理消费失败。
   • Custom Attributes 提供扩展能力。
   • Load 承载实际业务数据。
5. Message 有严格的大小限制：默认 4MB。避免大消息是关键设计原则，应通过拆分或外链方式处理大数据。
6. 最佳实践：
   • 合理使用 Tag 进行消息过滤。
   • 利用 Keys 进行消息追踪。
   • 遵守不可变性，通过 buildFrom 创建新消息。
   • 绝对不要发送超过 4MB 的消息，采用拆分或外链方案。

简而言之，Message 是 RocketMQ 的“信封”和“信件”本身。理解其结构、属性、约束和最佳实践，对于设计高效、可靠、可维护的消息系统至关重要。记住：小消息、不可变、善用属性、规避大负载。

您可能感兴趣的文章:

• RocketMQMessageListener注解对rocketmq消息的消费实现机制
• RocketMQ broker 消息投递流程处理PULL_MESSAGE请求解析
• 关于springboot使用rocketmq RocketMQMessageListener参数问题