原文地址：https://mp.weixin.qq.com/s/43wwC4lp77m4foVPEgTRlA

十分钟入门 RocketMQ

http://jm.taobao.org/2017/01/12/rocketmq-quick-start-in-10-minutes/

Publish/Subscribe

发布订阅是消息中间件的最基本功能，也是相对于传统 RPC 通信而言。

Message Priority

规范中描述的优先级是指在一个消息队列中，每条消息都有不同的优先级，一般用整数来描述，优先级高的消息先投递，如果消息完全在一个内存队列中，那么在投递前可以按照优先级排序，令优先级高的先投递。
由于 RocketMQ 所有消息都是持久化的，所以如果按照优先级来排序，开销会非常大，因此 RocketMQ 没有特意支持消息优先级，但是可以通过变通的方式实现类似功能，即单独配置一个优先级高的队列，和一个普通优先级的队列， 将不同优先级发送到不同队列即可。

Message Order

消息有序指的是一类消息消费时，能按照发送的顺序来消费。
例如：一个订单产生了3条消息，分别是订单创建，订单付款，订单完成。消费时，要按照这个顺序消费才能有意义。但是同时订单之间是可以并行消费的。
RocketMQ 可以严格的保证消息有序。

Message Filter

• Broker端消息过滤
  在 Broker 中，按照 Consumer 的要求做过滤，优点是减少了对于 Consumer 无用消息的网络传输。
  缺点是增加了 Broker 的负担，实现相对复杂。
• Consumer 端消息过滤
  这种过滤方式可由应用完全自定义实现，但是缺点是很多无用的消息要传输到Consumer端。

Message Persistence

消息中间件通常采用的几种持久化方式：

1. 持久化到数据库，例如Mysql。
2. 持久化到KV存储，例如levelDB、伯克利DB等KV存储系统。
3. 文件记录形式持久化，例如Kafka，RocketMQ
4. 对内存数据做一个持久化镜像，例如beanstalkd，VisiNotify

Message Reliablity

影响消息可靠性的几种情况：

1. Broker正常关闭
2. Broker异常Crash
3. OS Crash
4. 机器掉电，但是能立即恢复供电情况。
5. 机器无法开机（可能是cpu、主板、内存等关键设备损坏）
6. 磁盘设备损坏。

(1)、(2)、(3)、(4)四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）。

Low Latency Messaging

在消息不堆积情况下，消息到达 Broker 后，能立刻到达 Consumer。
RocketMQ 使用长轮询 Pull 方式，可保证消息非常实时，消息实时性不低于 Push。

At least Once

是指每个消息必须投递一次。
RocketMQ Consumer 先 Pull 消息到本地，消费完成后，才向服务器返回 ack，如果没有消费一定不会 ack 消息，所以 RocketMQ 可以很好的支持此特性。

Exactly Only Once

• 1. 发送消息阶段，不允许发送重复的消息。
• 2. 消费消息阶段，不允许消费重复的消息。

只有以上两个条件都满足情况下，才能认为消息是“Exactly Only Once”，而要实现以上两点，在分布式系统环境下，不可避免要产生巨大的开销。所以 RocketMQ 为了追求高性能，并不保证此特性，要求在业务上进行去重，也就是说消费消息要做到幂等性。

Broker的 Buffer 满了怎么办？

Broker 的 Buffer 通常指的是 Broker 中一个队列的内存 Buffer 大小，这类 Buffer 通常大小有限，如果Buffer满了以后怎么办？
RocketMQ 没有内存 Buffer 概念，RocketMQ 的队列都是持久化磁盘，数据定期清除。

对于此问题的解决思路，RocketMQ 同其他 MQ 有非常显著的区别，RocketMQ 的内存 Buffer 抽象成一个无限长度的队列，不管有多少数据进来都能装得下，这个无限是有前提的，Broker 会定期删除过期的数据，例如 Broker 只保存3天的消息，那么这个 Buffer 虽然长度无限，但是 3 天前的数据会被从队尾删除。

回溯消费

回溯消费是指 Consumer 已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker 在向 Consumer 投递成功消息后，消息仍然需要保留。
并且重新消费一般是按照时间维度，例如由于 Consumer 系统故障，恢复后需要重新消费1小时前的数据，那么 Broker 要提供一种机制，可以按照时间维度来回退消费进度。
RocketMQ支持按照时间回溯消费，时间维度精确到毫秒，可以向前回溯，也可以向后回溯。

消息堆积

消息中间件的主要功能是异步解耦，还有个重要功能是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的消息堆积能力。

定时消息

定时消息是指消息发到 Broker 后，不能立刻被 Consumer 消费，要到特定的时间点或者等待特定的时间后才能被消费。
如果要支持任意的时间精度，在 Broker 层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。
RocketMQ支持定时消息，但是不支持任意时间精度，支持特定的level，例如定时5s，10s，1m等。

消息重试

Consumer 消费消息失败后，要提供一种重试机制，令消息再消费一次。Consumer 消费消息失败通常可以认为有以下几种情况：

• 由于消息本身的原因，例如反序列化失败，消息数据本身无法处理（例如话费充值，当前消息的手机号被注销，无法充值）等。这种错误通常需要跳过这条消息，再消费其他消息，而这条失败的消息即使立刻重试消费，99%也不成功，所以最好提供一种定时重试机制，即过10s秒后再重试。
• 由于依赖的下游应用服务不可用，例如 DB 连接不可用，外系统网络不可达等。遇到这种错误，即使跳过当前失败的消息，消费其他消息同样也会报错。这种情况建议应用 sleep 30s，再消费下一条消息，这样可以减轻 Broker 重试消息的压力。

RocketMQ 是什么？

RocketMQ具有以下特点：

• 是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式特点。
• Producer、Consumer、队列都可以分布式。
• Producer 向一些队列轮流发送消息，队列集合称为 Topic
  • Consumer 如果做广播消费，则一个 Consumer 实例消费这个 Topic 对应的所有队列
  • Consumer 如果做集群消费，则多个 Consumer 实例平均消费这个 Topic 对应的队列集合。
• 能够保证严格的消息顺序
• 提供丰富的消息拉取模式
• 高效的订阅者水平扩展能力
• 实时的消息订阅机制
• 亿级消息堆积能力
• 较少的依赖

RocketMQ 物理部署结构

如上图所示， RocketMQ的部署结构有以下特点：

• Name Server 是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
• Broker 部署相对复杂，Broker 分为 Master 与 Slave。一个 Master 可以对应多个 Slave，但是一个 Slave 只能对应一个 Master，Master 与 Slave 的对应关系通过指定相同的 BrokerName，不同的 BrokerId 来定义，BrokerId 为 0 表示 Master，非 0 表示 Slave。Master 也可以部署多个。每个 Broker 与 Name Server集群中的所有节点建立长连接，定时注册 Topic 信息到所有 Name Server。
• Producer 与 Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从 Name Server 取 Topic 路由信息，并向提供 Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。Producer 完全无状态，可集群部署。
• Consumer 与 Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从 Name Server 取 Topic 路由信息，并向提供 Topic 服务的 Master、Slave 建立长连接，且定时向 Master、Slave 发送心跳。Consumer 既可以从 Master 订阅消息，也可以从 Slave 订阅消息，订阅规则由 Broker 配置决定。

RocketMQ 数据存储结构

RocketMQ 采取了一种数据与索引分离的存储方法。有效降低文件资源、IO资源，内存资源的损耗。即便是阿里这种海量数据，高并发场景也能够有效降低端到端延迟，并具备较强的横向扩展能力。

事务消息

以购物场景为例，张三购买物品，账户扣款 100 元的同时，需要保证在下游的会员服务中给该账户增加 100 积分。由于数据库私有，所以导致在实际的操作过程中会出现很多问题，比如先发送消息，可能会因为扣款失败导致账户积分无故增加，如果先执行扣款，则有可能因服务宕机，导致积分不能增加。
无论是先发消息还是先执行本地事务，都有可能导致出现数据不一致的结果。

事务消息的本质就是为了解决此类问题，解决本地事务执行与消息发送的原子性问题。
目前，事务消息在多种分布式消息中间件种均有实现，但是其实现方式思路却各有不同。

传统事务消息实现

传统事务消息实现，一种思路是依赖于 AMQP 协议用来确保消息发送成功，AMQP 模式下需要在发送在发送事务消息时进行两阶段提交，首先进行 tx_select 开启事务，然后再进行消息发送，最后进行消息的 commit 或者是 rollback。
这个过程可以保证在消息发送成功的同时本地事务也一定成功执行，但事务粒度不好控制，而且会导致性能急剧下降，同时依然无法解决本地事务执行与消息发送的原子性问题。

还有另外一种思路，就是通过保证多条消息的同时可见性来保证事务一致性。但是此类消息事务实现机制更多的是用到 consume-transform-produce 场景中，其本质还是用来保证消息自身事务，并没有把外部事务包含进来。

RocketMQ 事务消息

RocketMQ 事务消息设计则主要是为了解决 Producer 端的消息发送与本地事务执行的原子性问题。
RocketMQ 的设计中 broker 与 producer 端的双向通信能力，使得 broker 天生可以作为一个事务协调者存在；而 RocketMQ 本身提供的存储机制，则为事务消息提供了持久化能力；RocketMQ 的高可用机制以及可靠消息设计，则为事务消息在系统在发生异常时，依然能够保证事务的最终一致性达成。

RocketMQ 事务消息设计

事务消息作为一种异步确保型事务， 将两个事务分支通过 MQ 进行异步解耦，RocketMQ 事务消息的设计流程同样借鉴了两阶段提交理论，整体交互流程如下图所示：

image

• 事务发起方（即消息发送者）首先发送 prepare 消息到 MQ。
• 事务发起方（即消息发送者）在发送 prepare 消息成功后执行本地事务。
• 根据本地事务执行结果发送 commit 或者是 rollback 给 MQ。
  • 如果消息是 rollback，MQ 将删除该 prepare 消息不进行下发。
  • 如果消息是 commit，MQ 将会把这个消息发送给 consumer 端。
• 如果执行本地事务过程中，执行端挂掉，或者超时，导致 MQ 收不到任何的消息（不知道是该 commit 还是该 rollback），RocketMQ 会定期扫描消息集群中的事务消息，这时候发现了某个 prepare 消息还不知道该怎么处理，它会向消息发送者确认，所以消息发送者需要实现一个 check 接口，RocketMQ 会根据消息发送者设置的策略来决定是 rollback 还是继续 commit。这样就保证了消息发送与本地事务同时成功或同时失败。
• Consumer 端的消费成功机制由 MQ 保证。

image

RocketMQ 事务消息实现

在具体实现上，RocketMQ 通过使用 Half Topic 以及 Operation Topic 两个内部队列来存储事务消息推进状态，如下图所示：

image

其中，Half Topic 对应队列中存放着 prepare 消息，Operation Topic 对应的队列则存放了 prepare message 对应的 commit/rollback 消息，消息体中则是 prepare message 对应的 offset，服务端通过比对两个队列的差值来找到尚未提交的超时事务，进行回查。

从用户侧来说，用户需要分别实现本地事务执行以及本地事务回查方法，因此只需关注本地事务的执行状态即可；而在 service 层，则对事务消息的两阶段提交进行了抽象，同时针对超时事务实现了回查逻辑，通过不断扫描当前事务推进状态，来不断反向请求 Producer 端获取超时事务的执行状态，在避免事务挂起的同时，也避免了 Producer 端的单点故障。
而在存储层，RocketMQ 通过 Bridge 封装了与底层队列存储的相关操作，用以操作两个对应的内部队列，用户也可以依赖其他存储介质实现自己的 service，RocketMQ 会通过 ServiceProvider 加载进来。

从上述事务消息设计中可以看到，RocketMQ 事务消息较好的解决了事务的最终一致性问题，事务发起方仅需要关注本地事务执行以及实现回查接口给出事务状态判定等实现，而且在上游事务峰值高时，可以通过消息队列，避免对下游服务产生过大压力。