RocketMQ 入门

RocketMQ 是一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。

官方文档：

https://github.com/apache/rocketmq
https://rocketmq.apache.org/zh/

基本概念

主题（Topic）

主题（Topic）可以看做消息的归类，表示一类消息的集合，它是消息的第一级类型，用于标识同一类业务逻辑的消息。每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。

主题内部由多个队列组成，消息的存储和水平扩展能力最终是由队列实现的；并且针对主题的所有约束和属性设置，最终也是通过主题内部的队列来实现。

消息标签（MessageTag）

消息标签主要用于同一主题下区分不同类型的消息，如果把 Topic 比作一级分类，那 tag 就是二级分类。

Topic: 货物
	tag=上海
	tag=北京
	tag=天津

# 消费者可以根据Tag实现对不同子主题的不同消费逻辑
1. topic=货物 tag=上海
2. topic=货物 tag=上海|北京

队列（MessageQueue）

队列是 RocketMQ 存储消息的物理实体，一个 Topic 中可以包含多个 Queue，每个 Queue 中存放的就是该Topic的消息。一个 Topic 的 Queue 也被称为一个 Topic中 消息的分区（Partition）。

一个Topic的Queue中的消息只能被一个消费者组中的一个消费者消费。一个Queue中的消息不允许同一个消费者组中的多个消费者同时消费。

消息标识（MessageId/Key）

RocketMQ 中每个消息拥有唯一的 MessageId，且可以携带具有业务标识的 Key，以方便对消息的查询。不过需要注意的是，MessageId 有两个：在生产者发送消息时会自动生成一个 MessageId（msgId)，当消息到达 Broker 后，Broker 也会自动生成一个 MessageId(offsetMsgId)。msgId、offsetMsgId 与 key 都称为消息标识。
 msgId ：由producer端生成，其生成规则为：producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode +当前时间 + AutomicInteger自增计数器
offsetMsgId：由broker端生成，其生成规则为：brokerIp + 物理分区的offset（Queue中的偏移量）
key：由用户指定的业务相关的唯一标识

RocketMQ 核心概念

Name Server

NameServer 是 RocketMQ Broker与Topic路由的注册中心，支持Broker的动态注册与发现。在启动服务的时候，NameServer 需要先于 Broker 启动，每个 Broker 在启动的时候会到 NameServer 注册。
每个 NameServer 中都保存着 Broker 集群的整个路由信息和用于客户端查询的队列信息。Producer 和 Conumser 通过 NameServer可 以获取整个 Broker 集群的路由信息，从而进行消息的投递和消费。

路由注册

NameServer 被设计成几乎无状态的，可以横向扩展，节点之间相互之间无通信，通过部署多台机器来标记自己是一个伪集群。在 Broker 节点启动时，轮询NameServer 列表，与每个 NameServer 节点建立长连接，发起注册请求。在 NameServer 内部维护着一个 Broker 列表，用来动态存储 Broker 的信息。

优点：NameServer集群搭建简单，扩容简单。
缺点：对于Broker，必须明确指出所有 NameServer 地址。否则未指出的将不会去注册。也正因为如此，NameServer 并不能随便扩容。因为，若Broker不重新配置，新增的NameServer对于Broker来说是不可见的，其不会向这个NameServer进行注册。
Broker 每30s会发送一次心跳包给 NameServer，心跳包中包含BrokerId、Broker地址(IP+Port)、Broker名称、Broker所属集群名称等。NameServer在接收到心跳包后，会更新心跳时间戳，记录这个Broker的最新存活时间。

路由剔除

如果发生网络阻塞或者 Broker关机等情况，NameServer没有收到Broker的心跳，NameServer可能会将其从Broker列表中剔除。
NameServer 的定时任务每隔10s会扫描一次 Broker 表，查看每一个Broker的最新心跳时间戳距离当前时间是否超过 120 秒，如果超过，则会判定Broker失效，然后将其从Broker列表中剔除。

Broker升级，需要停掉Broker的工作，需要怎么做？

1. 将Broker的读写权限禁掉。一旦client(Consumer或Producer)向broker发送请求，都会收到broker的NO_PERMISSION响应，然后client会进行对其它Broker的重试。
2. Broker没有流量后，再关闭它，实现Broker从NameServer的移除。

路由发现

路由发现采用的是Pull模型。当Topic路由信息出现变化时，NameServer不会主动推送给客户端，而是客户端定时拉取主题最新的路由。默认客户端每 30 秒会拉取一次最新的路由。

NameServer选择策略（Producer与Consumer）

会生产一个随机数，然后再与NameServer节点数量取模，此时得到的就是所要连接的节点索引，然后就会进行连接。如果连接失败，则会采用round-robin（轮询）策略，逐个尝试着去连接其它节点。

Broker

Broker 是消息存储中心，主要负责存储消息、转发消息。存储与消息相关的元数据，包括用户组、消费进度偏移量、队列信息等。
Broker 分为 Master 与 Slave 两种，从物理结构上看 Broker 的集群部署方式有四种：单 Master 、多 Master 、多 Master 多 Slave（同步刷盘）、多 Master多 Slave（异步刷盘）。

Broker的组成

1. Client Manager：客户端管理器。负责接收、解析客户端(Producer/Consumer)请求，管理客户端。例如，维护Consumer的Topic订阅信息
2. Store Service：存储服务。提供方便简单的API接口，处理消息存储到物理硬盘和消息查询功能。
3. HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
4. Index Service：索引服务。根据特定的Message key，对投递到Broker的消息进行索引服务，同时也提供根据Message Key对消息进行快速查询的功能。

Broker的集群部署

Broker一般都是以集群形式出现的，各个节点存放着相同Topic的不同Queue。可以通过将集群每个Broker集群节点进行横向扩展，即将Broker节点再建为一个HA集群，解决单点问题。

Master负责处理读写操作请求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slave则会自动切换为Master去工作。一个Master可以包含多个Slave，但一个Slave只能隶属于一个Master。Master与Slave 的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。BrokerId为 0 表示Master，非 0 表示Slave。

Broker集群模式

1. 单Master
   只有一个broker（其本质上就不能称为集群）。这种方式也只能是在测试时使用，生产环境下不能使用，因为存在单点问题。
2. 多Master
   broker集群仅由多个master构成，不存在Slave。同一Topic的各个Queue会平均分布在各个master节点上。

   优点：配置简单，单个Master宕机或重启维护对应用无影响
   缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅（不可消费），消息实时性会受到影响。

3. 多Master多Slave模式-异步复制

broker集群由多个master构成，每个master又配置了多个slave,master与slave的关系是主备关系，即master负责处理消息的读写请求，而slave仅负责消息的备份与master宕机后的角色切换。

该模式的最大特点之一是，当master宕机后slave能够自动切换为master。不过由于slave从master的同步具有短暂的延迟（毫秒级），所以当master宕机后，这种异步复制方式可能会存在少量消息的丢失问题。

4. 多Master多Slave模式-同步双写
   该模式是多Master多Slave模式的同步复制实现。所谓同步双写，指的是消息写入master成功后，master会等待slave同步数据成功后才向producer返回成功ACK，即master与slave都要写入成功后才会返回成功ACK，也即双写。该模式与异步复制模式相比，优点是消息的安全性更高，不存在消息丢失的情况。但单个消息的RT略高，从而导致性能要略低（大约低10%）。

该模式存在一个大的问题：对于目前的版本，Master宕机后，Slave不会自动切换到Master。
复制策略
复制策略是Broker的Master与Slave间的数据同步方式。分为同步复制与异步复制：

• 同步复制：消息写入master后，master会等待slave同步数据成功后才向producer返回成功ACK
• 异步复制：消息写入master后，master立即向producer返回成功ACK，无需等待slave同步数据成功

刷盘策略
刷盘策略指的是broker中消息的落盘方式，即消息发送到broker内存后消息持久化到磁盘的方式。分为同步刷盘与异步刷盘。

• 同步刷盘：当消息持久化到broker的磁盘后才算是消息写入成功。
• 异步刷盘：当消息写入到broker的内存后即表示消息写入成功，无需等待消息持久化到磁盘。

生产者（Producer）

消息发布者，负责生产并发送消息至 Topic。RocketMQ 支持同步发送、异步发送、单向发送。

消费者（Consumer）

消息订阅者，负责从 Topic 接收并消费消息。消费者从brokers那里拉取信息并将其输入应用程序。

两种类型的消费者：

1. Pull：Pull型消费者主动地从brokers那里拉取信息。只要批量拉取到消息，用户应用程序就会启动消费过程
2. Push：Push型消费者封装消息的拉取、消费进度和维护内部的其他工作，将一个在消息到达时执行的回调接口留给终端用户来实现。

工作流程

1. 启动NameServer，NameServer启动后开始监听端口，等待Broker、Producer、Consumer连接。
2. 启动Broker时，Broker会与所有的NameServer建立并保持长连接，然后每 30 秒向NameServer定时发送心跳包。
3. 发送消息前，可以先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，当然，在创建Topic时也会将Topic与Broker的关系写入到NameServer中。不过，这步是可选的，也可以在发送消息时自动创建Topic。
4. Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取路由信息，即当前发送的Topic消息的Queue与Broker的地址（IP+Port）的映射关系。然后根据算法策略从队选择一个Queue，与队列所在的Broker建立长连接从而向Broker发消息。当然，在获取到路由信息后，Producer会首先将路由信息缓存到本地，再每 30 秒从NameServer更新一次路由信息。
5. Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取其所订阅Topic的路由信息，然后根据算法策略从路由信息中获取到其所要消费的Queue，然后直接跟Broker建立长连接，开始消费其中的消息。Consumer在获取到路由信息后，同样也会每 30 秒从NameServer更新一次路由信息。不过不同于Producer的是，Consumer还会向Broker发送心跳，以确保Broker的存活状态。