消息队列与RocketMQ总结

一个 Topic 有多个 Partition，不同 Partition 可以分布到不同的 Broker 上

可能造成消费者来不及处理消息

更适合 Kafka，按照自己的节奏消费消息

RocketMQ就是此模型

master结点崩溃

一些salve结点崩溃

所有slave结点崩溃

用于标识同一类业务逻辑的消息

队列作为主题的组成单元, 是消息存储的实际容器, 一个主题内包含一个或多个队列, 消息实际存储在主题的各队列内

每个主题只允许发送一种类型的消息。主题支持的消息类型：普通消息、顺序消息、延时消息、事务消息

将相同业务域内同一功能属性的消息划分为同一主题

拆分方式

通过主题隔离业务, 不同业务逻辑的消息建议使用不同的主题

存储顺序性、流式操作语义

遵循少用够用原则

需要增加队列实现物理节点负载均衡

消息是 RocketMQ 中的最小数据传输单元. 生产者将业务数据的负载和拓展属性包装成消息发送到 RocketMQ 服务端, 服务端按照相关语义将消息投递到消费端进行消费

消息不可变性

消息持久化

大负载按照固定大小做报文拆分, 或者结合文件存储等方法进行传输

生产者是 RocketMQ 系统中用来构建并传输消息到服务端的运行实体

生产者和主题是多对多的关系, 支持同一个生产者向多个主题发送消息；如果有需要发送消息到多个主题的场景, 无需为每个主题都创建一个生产者

消费者分组是 RocketMQ 系统中承载多个消费行为一致的消费者的负载均衡分组。通过消费者分组内初始化多个消费者实现消费性能的水平扩展以及高可用容灾

订阅关系

投递顺序性

消费重试策略

消费者投递顺序一致

消费者业务类型一致

消费者是 RocketMQ 中用来接收并处理消息的运行实体

大部分场景下, 单一进程内同一个消费分组只需要初始化唯一的一个消费者即可

订阅关系是 RocketMQ 系统中消费者获取消息, 处理消息的规则和状态配置

订阅关系由消费者分组动态注册到服务端系统, 并在后续的消息传输中按照订阅关系定义的过滤规则进行消息匹配和消费进度维护

消息过滤规则

消费状态

不同消费者分组对于同一个主题的订阅相互独立

同一个消费者分组对于不同主题的订阅也相互独立

订阅关系关联了过滤规则, 消费进度等元数据和相关配置, 同时系统需要保证消费者分组下的所有消费者的消费行为, 消费逻辑, 负载策略等一致, 整体运算逻辑比较复杂

同一ConsumerGroup下的Consumer实例订阅的Topic不同(3.x, 4.x SDK适用)

同一ConsumerGroup下的Consumer实例订阅的Topic相同,但订阅的Tag不一致

初始化

待消费

消费中

消费提交

消息删除

离线日志收集

若存储系统异常重启, 可能会导致定时消息投递出现一定延迟.

相比普通消息，多了一个【定时中】状态

会造成系统压力过大, 导致消息分发延迟, 影响定时精度

比如订单超时未支付进行取消的场景

支持消费者按照发送消息的先后顺序获取消息

强调多条消息间的先后顺序关系

顺序消息的顺序关系通过消息组(MessageGroup)判定和识别, 发送顺序消息时需要为每条消息设置归属的消息组, 相同消息组的多条消息之间遵循先进先出的顺序关系, 不同消息组, 无消息组的消息之间不涉及顺序性

通过生产者和服务端的协议保障单个生产者串行地发送消息, 并按序存储和持久化

可以使用 hash 法将同一个业务 ID 的消息发送到同一个队列中

顺序消息发送必须要设置消息组

相同消息组的消息按照先后顺序被存储在同一个队列.

不同消息组的消息可以混合在同一个队列中, 且不保证连续.

按照投递的顺序进行消费

顺序消息投递仅在重试次数限定范围内, 即一条消息如果一直重试失败, 超过最大重试次数后将不再重试, 跳过这条消息消费, 不会一直阻塞后续消息处理。顺序消息消费失败进行消费重试时, 为保障消息的顺序性, 后续消息不可被消费, 必须等待前面的消息消费完成后才能被处理

一般建议的消息组设计会采用订单 ID, 用户 ID 作为顺序参考, 即同一个终端用户的消息保证顺序, 不同用户的消息无需保证顺序

建议将业务以消息组粒度进行拆分

上游源端数据库按需执行增删改操作, 将二进制操作日志作为消息

1. 生产者将消息发送至 RocketMQ 服务端.
2. RocketMQ 服务端将消息持久化成功之后, 向生产者返回 Ack 确认消息已经发送成功, 此时消息被标记为 "暂不能投递", 这种状态下的消息即为半事务消息.
3. 生产者开始执行本地事务逻辑.
4. 生产者根据本地事务执行结果向服务端提交二次确认结果(Commit 或是 Rollback), 服务端收到确认结果后处理逻辑如下:
二次确认结果为 Commit: 服务端将半事务消息标记为可投递, 并投递给消费者.
二次确认结果为 Rollback: 服务端将回滚事务, 不会将半事务消息投递给消费者.
5. 在断网或者是生产者应用重启的特殊情况下, 若服务端未收到发送者提交的二次确认结果, 或服务端收到的二次确认结果为 Unknown 未知状态, 经过固定时间后, 服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查. 服务端回查的间隔时间默认值: 60秒.
6. 生产者收到消息回查后, 需要检查对应消息的本地事务执行的最终结果.
7. 生产者根据检查到的本地事务的最终状态再次提交二次确认, 服务端仍按照步骤 4 对半事务消息进行处理.

相比普通消息，多了事务待提交、消息回滚、提交待消费的状态

事务消息保证本地主分支事务和下游消息发送事务的一致性, 但不保证消息消费结果和上游事务的一致性

RocketMQ 事务消息为最终一致性, 即在消息提交到下游消费端处理完成之前, 下游分支和上游事务之间的状态会不一致. 因此, 事务消息仅适合接受异步执行的事务场景

RocketMQ 支持在事务提交阶段异常的情况下发起事务回查, 保证事务一致性. 但生产者应该尽量避免本地事务返回未知结果. 大量的事务检查会导致系统性能受损, 容易导致事务处理延迟

消息回查时, 对于正在进行中的事务不要返回 Rollback 或 Commit 结果, 应继续保持 Unknown 的状态

如何保证核心业务和多个下游业务的执行结果完全一致, 是分布式事务需要解决的主要问题

基于 RocketMQ 分布式事务消息: 支持最终一致性

基于 RocketMQ 实现的分布式事务消息功能, 在普通消息基础上, 支持二阶段的提交能力. 将二阶段提交和本地事务绑定, 实现全局提交结果的一致性

部分节点异常是否影响消息发送?
请求重试是否会阻塞业务调用?
请求重试会带来什么不足?

定义

生产者初始化最大重试次数，后续发送消息失败会一直重试，直到最大次数

除服务端返回系统流控错误场景, 其他触发条件触发重试后, 均会立即进行重试, 无等待间隔

若由于服务端返回流控错误触发重试, 系统会按照指数退避策略进行延迟重试

需要合理评估每次调用请求的超时时间以及最大重试次数, 避免影响全链路的耗时

RocketMQ 客户端内置的发送请求重试机制并不能保证消息发送一定成功，业务方需要做好兜底

系统在什么情况下会触发流控?
触发流控时客户端行为是什么?
应该如何避免触发流控, 以及如何应对突发流控?

消息流控指的是系统容量或水位过高, RocketMQ 服务端会通过快速失败返回流控错误来避免底层资源承受过高压力.

存储压力大

服务端请求任务排队溢出

系统触发消息发送流控时, 客户端会收到系统限流错误和异常

客户端收到系统流控错误码后, 会根据指数退避策略进行消息发送重试.

通过监控避免触发流控

基于 SDK 内部的典型 Reactor 线程模型实现

SDK 内置了一个长轮询线程, 先将消息异步拉取(pull)到 SDK 内置的缓存队列中, 再分别提交到消费线程中, 触发监听器执行本地消费逻辑

所有消息必须以同步方式进行消费处理, 并在监听器接口结束时返回调用结果, 不允许再做异步化分发

PushConsumer 严格限制了消息同步处理及每条消息的处理超时时间

消息的获取, 消费状态提交以及消费重试都是通过消费者业务逻辑主动发起调用完成。由业务逻辑按需调用接口获取消息, 然后分发给业务线程处理消息, 最后按照处理的结果调用提交接口, 返回服务端当前消息的处理结果

SimpleConsumer 提供原子接口, 用于消息获取和提交消费结果, 相对于 PushConsumer 方式更加灵活

比如交易系统会产生订单、支付、物流相关的消息。不同的业务线可以对消息进行过滤，只消费自己感兴趣的消息

生产者

消费者

服务端

生产者发送消息时可以自定义消息属性, 每个属性都是一个自定义的键值对(Key-Value). 每条消息支持设置多个属性

业务消息的拆分可以基于主题进行筛选, 也可以基于主题内消息的 Tag 标签及属性进行筛选

消息消费处理的容灾策略

消息消费的顺序性机制

消息分配的水平拆分策略

每个消费者分组只初始化唯一一个消费者, 每个消费者可消费到消费者分组内所有的消息, 各消费者分组都订阅相同的消息, 以此实现单客户端级别的广播一对多推送效果. 该方式一般可用于网关推送, 配置推送等场景

每个消费者分组下初始化了多个消费者, 这些消费者共同分担消费者分组内的所有消息, 实现消费者分组内流量的水平拆分和均衡负载. 该方式一般可用于微服务解耦场景

消息粒度负载均衡: PushConsumer 和 SimpleConsumer 默认负载策略

队列粒度负载均衡: PullConsumer 默认负载策略

之前的版本都是队列粒度的负载均衡

消费分摊更均衡

对非对等消费者更友好

队列分配运维更方便

同一队列内的消息离散地分布于多个消费者, 适用于绝大多数在线事件处理的场景

同一消费者分组内的多个消费者将按照队列粒度消费消息, 即每个队列仅被一个消费者消费

队列粒度负载均衡策略分配粒度较大, 不够灵活. 但该策略在流式处理场景下有天然优势, 能够保证同一队列的消息被相同的消费者处理, 对于批量处理, 聚合处理更友好

适用于流式计算, 数据聚合等需要明确对消息进行聚合, 批处理的场景

无论是消息粒度负载均衡策略还是队列粒度负载均衡策略, 在消费者上线或下线, 服务端扩缩容等场景下, 都会触发短暂的重新负载均衡动作. 此时可能会存在短暂的负载不一致情况, 出现少量消息重复的现象. 因此, 需要在下游消费逻辑中做好消息幂等去重处理

消费者启动后从哪里开始消费消息?
消费者每次消费成功后如何标记消息状态, 确保下次不会再重复处理该消息?
某消息被指定消费者消费过一次后, 如果业务出现异常需要做故障恢复, 该消息能否被重新消费?

消息是按到达服务端的先后顺序存储在指定主题的多个队列中, 每条消息在队列中都有一个唯一的 Long 类型坐标, 这个坐标被定义为消息位点

任意一个消息队列在逻辑上都是无限存储, 即消息位点会从 0 到 Long.MAX 无限增加。RocketMQ 定义队列中最早一条消息的位点为最小消息位点(MinOffset); 最新一条消息的位点为最大消息位点(MaxOffset). 虽然消息队列逻辑上是无限存储, 但由于服务端物理节点的存储空间有限, RocketMQ 会滚动删除队列中存储最早的消息. 因此, 消息的最小消费位点和最大消费位点会一直递增变化

每个主题的队列都可以被多个消费者分组订阅. 若某条消息被某个消费者消费后直接被删除, 则其他订阅了该主题的消费者将无法消费该消息

RocketMQ 通过消费位点管理消息的消费进度. 每条消息被某个消费者消费完成后不会立即在队列中删除, RocketMQ 会基于每个消费者分组维护一份消费记录, 该记录指定消费者分组消费某一个队列时, 消费过的最新一条消息的位点, 即消费位点

当消费者客户端离线, 又再次重新上线时, 会严格按照服务端保存的消费进度继续处理消息. 如果服务端保存的历史位点信息已过期被删除, 此时消费位点向前移动至服务端存储的最小位点

每个消费组在每个队列上维护一个消费位置 offset, 因为在发布订阅模式中一般会涉及到多个消费者组, 而每个消费者组在每个队列中的消费位置都是不同的

消费位点初始值指的是消费者分组首次启动消费者消费消息时, 服务端保存的消费位点的初始值

RocketMQ 定义消费位点的初始值为消费者首次获取消息时, 该时刻队列中的最大消息位点. 相当于消费者将从队列中最新的消息开始消费

在首次上线时会从服务器中的最新消息开始消费, 也就是从队列的尾部开始消费; 再次重新启动后, 会从最后一次的消费位置继续消费

回溯消费是指已经消费成功的消息, 由于业务上需求需要重新消费

支持按照时间回溯消费, 时间维度精确到毫秒

初始消费位点不符合需求

消费堆积快速清理

业务回溯，纠正处理

对于回溯重置类场景, 重置后的历史消息大多属于存储冷数据, 可能会造成系统压力上升, 一般称为冷读现象

只能重置对消费者可见的消息, 不能重置定时中, 重试等待中的消息

生产者生产太快或消费者消费太慢

解决方法：增加消费者

业务处理失败, 且失败原因跟当前的消息内容相关

消费失败的原因不会导致连续性, 即当前消息消费失败是一个小概率事件, 不是常态化的失败, 后面的消息大概率会消费成功

如何保证业务完整处理消息

系统异常时处理中的消息状态如何恢复

消费失败

消息处理超时

已就绪状态

处理中状态

待重试状态

提交状态

死信状态

与 PushConsumer 一致，只是少了 待重试状态

消息在服务端中的存储以什么维度为判定条件? 消息存储以什么粒度进行管理? 消息存储超过限制后如何处理? 这些问题都是由消息存储和过期清理机制来定义的

RocketMQ 使用存储时长作为消息存储的依据, 即每个节点对外承诺消息的存储时长. 在存储时长范围内的消息都会被保留, 无论消息是否被消费; 超过时长限制的消息则会被清理掉

按照服务端节点粒度管理存储时长而非队列或主题

RocketMQ 统一管理消息的存储时长, 无论消息是否被消费

当消费者不在线或消息消费异常时, 会造成队列中大量消息堆积, 且该现象暂时无法有效控制. 若此时按照消费状态考虑将未消费的消息全部保留, 则很容易导致存储空间不足, 进而影响到新消息的读写速度

默认存储在本地磁盘中

Tag的使用

Keys的使用

一定要打印日志

重试方式没有集成到 MQ 客户端内部做, 而是要求应用自己去完成

RocketMQ 无法避免消息重复(Exactly-Once), 所以如果业务对消费重复非常敏感, 务必要在业务层面进行去重处理

首先需要确定消息的唯一键, 可以是 msgId, 也可以是消息内容中的唯一标识字段, 例如订单 Id 等. 在消费之前判断唯一键是否在关系数据库中存在. 如果不存在则插入, 并消费, 否则跳过

也可以使用写入 Redis 来保证, 因为 Redis 的 key 和 value 就是天然支持幂等的

提高消费并行度

批量方式消费

重置位点跳过非重要消息

优化每条消息消费过程

如果消息量较少, 建议在消费入口方法打印消息, 消费耗时等, 方便后续排查问题

3000ms

不超过 4M

3次

默认 16 次

Controller嵌入NameServer部署

Controller独立部署

Expoter 通过 MQAdminExt 向 MQ 集群请求数据, 请求到的数据通过 MetricService 规范化成 Prometheus 需要的格式, 然后通过 /metics 接口暴露给 Promethus

Metric 类位于 org.apache.rocketmq.expoter.model.metrics 包下, 实质上是一些实体类, 每个实体类代表一类指标, 总共 14 个 Metric 类

RocketMQ-exporter 项目和 Prometheus 相当于服务器和客户端的关系, RocketMQ-exporter 项目引入了 Prometheus 的 client 包, 该包中规定了需要获取的信息的类型即项目中的 MetricFamilySamples 类, Prometheus 向 expoter 请求 metrics, expoter 将信息封装成相应的类型之后返回给 Prometheus