RocketMQ 原理与基础

导读：RocketMQ 主要用于解决系统内部耦合度过高、瞬间流量冲击等问题。它通过异步通信机制，实现了高性能、高可靠的消息传递。

---

RocketMQ 简介

RocketMQ 是一个 分布式消息中间件，可以看作是 “消息队列 + 分布式系统” 的结合体。

• 作为消息队列：遵循“发-存-收”模型（Producer-Broker-Consumer）。
• 作为分布式系统：由服务端（Broker）、客户端（Producer/Consumer）和注册中心（NameServer）组成。

核心工作流程：

1. Broker 注册：启动时向所有 NameServer 注册，并每 30s 发送一次心跳。
2. Producer 发送：从 NameServer 获取路由，根据负载均衡算法选择一台 Broker 发送。
3. Consumer 消费：从 NameServer 获取路由，主动拉取（Pull）或被动接收（Push）消息进行消费。

---

一、RocketMQ 的核心组成

RocketMQ 由四个核心部分组成，为了保证高可用，通常每一部分都是集群部署的。

组件	角色	核心功能
NameServer	注册中心	发现服务，维护 Topic 路由信息，无状态服务器。
Broker	消息中转	存储消息、转发消息，维护 Consumer Queue 和 CommitLog。
Producer	生产者	发送消息，支持同步、异步、单向发送。
Consumer	消费者	处理消息，支持集群消费和广播消费。

1. NameServer

• 特性：无状态，结点间不通信。
• 职责：Broker 定时注册，Producer/Consumer 定时获取路由。

2. Broker

• 内部构造：CommitLog（真实存储）+ Consumer Queue（消息索引）。
• 通信：基于 Netty 实现长连接心跳。

---

二、消息的逻辑结构

1. Message (消息)

传输信息的最小载体。必须包含 Topic，可选包含 Tag 和 Keys（业务唯一标识）。

2. Topic (主题)

消息的第一级分类。一个 Topic 可以有多个生产者和消费者。

3. Tag (标签)

消息的第二级分类。用于在同一业务模块下区分不同用途的消息（如：交易 Topic 下区分“支付”和“退款”Tag）。

4. Consumer Group (消费组)

订阅者的逻辑聚合。

• 集群消费：组内成员竞争消费（一条消息只被组内一个成员处理）。
• 负载均衡：自动将 Message Queue 分配给组内成员。

5. Message Queue & Offset

• Queue：Topic 的物理分区，用于并行消费。
• Offset：消费进度的下标，记录消费到了哪一条。

三、消息队列模型

1. 队列模型

传统的“点对点”模型。多个消费者之间是竞争关系，每条消息只能被消费一次。

2. 发布/订阅模型

RocketMQ 采用的模型。允许多个消费组订阅同一个 Topic，每份订阅都能收到全量消息。

---

四、消息发送模式对比

特性	同步发送 (Sync)	异步发送 (Async)	单向发送 (OneWay)
阻塞	阻塞等待响应	不阻塞，回调处理	不阻塞，无响应
可靠性	极高	高（需处理回调）	低
性能	一般	高	极高
场景	重要通知、短信	耗时业务、转码	日志收集

---

五、消息消费重试机制

原理：如果消费者处理失败（抛出异常或返回 RECONSUME_LATER），消息会进入重试队列。

• 重试次数：默认 16 次。
• 时间间隔：阶梯式增长（1s, 5s, 10s... 直到 2h）。
• 死信队列 (DLQ)：16 次重试均失败后，消息进入死信队列，需人工干预。
• 要求：消费者必须实现幂等性，防止重复消费导致的业务错误。

---

六、消息过滤机制（Broker 端过滤）

RocketMQ 强大的地方在于它支持在 Broker 服务端 直接过滤消息，节省网络带宽。

1. Tag 过滤

最常用。基于 Tag 的 HashCode 过滤。

// 注解方式 引用的包是rocketmq-spring-boot-starter
@RocketMQMessageListener(topic = "TOPIC", selectorExpression = "TagA || TagB")
//原生MQ方式 如果引用的包是 rocketmq-client-java
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

2. SQL92 过滤

最灵活。支持逻辑运算（age > 18 AND sex = 'male'）。

• 前提：Broker 需开启 enablePropertyFilter=true。