消息中间件需要解决哪些问题？

阳光穿透心脏的1/2处 2022-04-25 01:06 144阅读 0赞

**目录**

1、发布订阅

2、消息优先级（Message Priority）

3、消息有序（Message Order）

4、消息过滤

5、消息持久化

6、消息可靠性（Message Reliablity）

7、低延迟消息传递

8、At least Once

9、Exactly Only Once

10、Broker 的 Buffer 满了怎么办？

11、回溯消费

12、消息堆积

13、分布式事务

14、定时消息

15、消息重试

--------------------

# 1、发布订阅 #

# 2、消息优先级（Message Priority） #

优先级是指在一个消息队列中每条消息都有不同的优先级，一般用整数来描述，优先级高的消息先投递，如果消息完全在一个内存队列中，那么在投递前可以按照优先级排序，令优先级高的先投递。

由于 RocketMQ 所有消息都是持久化的，所以如果按照优先级来排序，开销会非常大，因此RocketMQ 没有特意支持消息优先级但是可以通过变通的方式实现类似功能，即单独配置一个优先级高的队列，和一个普通优先级的队列，将不同优先级发送到不同队列即可。

# 3、消息有序（Message Order） #

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

RocketMQ 可以严格的保证消息有序。

# 4、消息过滤 #

（1）Broker 端消息过滤

在 Broker 中，按照 Consumer 的要求做过滤，优点是减少了对于 Consumer 无用消息的网络传输。

缺点：增加了 Broker 的负担，实现相对复杂。

（2）Consumer 端消息过滤

优点：过滤方式由应用完全自定义实现，比较灵活；

缺点：很多无用的消息要传输到 Consumer 端。

# 5、消息持久化 #

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

（1）持久化到数据库，例如 MySQL。

（2）持久化到 kv存储，例如 levelDB等。

（3）文件记录形式持久化，例如 Kafka、RocketMQ。

（4）对内存数据做一个持久化镜像，例如 beanstalkd、VisiNotify。

（1）（2）（3）三种持久化方式都具有将内存队列 Buffer 进行扩展的能力，（4）只是一个内存的镜像，作用是当 Broker 挂掉重启后仍然能将之前内存的数据恢复出来。

持久化的性能直接决定中间件的性能。

RocketMQ 参考了 Kafka 的持久化方式，充分利用 Linux 文件系统内存 cache 来提高性能。

# 6、消息可靠性（Message Reliablity） #

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

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

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

5、 6 属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。RocketMQ 在这两种情况下，通过异步复制，可保证 99% 的消息不丢，但是仍然会有极少数量的消息可能丢失。通过同步双写技术可以完全避免单点，同步双写势必会影响性能，适合对消息可靠性要求极高的场合，例如与 Money 相关的应用。

RocketMQ 从 3.0 版本开始支持同步双写。

# 7、低延迟消息传递 #

在消息不堆积情况下，消息到达 Broker 后，能立刻到达 Consumer。

RocketMQ 使用长轮询 Pull 方式，可保证消息非常实时，消息实时性不低于 Push。

# 8、At least Once #

是指每个消息必须投递一次。

RocketMQ Consumer 先 pull 消息到本地，消费完成后，才向服务器返回 ack，如果没有消费一定不会 ack 消息，所以 RocketMQ 可以很好的支持此特性。

# 9、Exactly Only Once #

（1）发送消息阶段，不允许发送重复的消息。

（2）消费消息阶段，不允许消费重复的消息。

同时满足这两个条件，才能认为消息是“Exactly Only Once”，而要实现以上两点，在分布式系统环境下，不可避免要产生巨大的开销，所以 RocketMQ 为了追求高性能，并不保证此特性，要求在业务上进行去重，也就是说消费消息要做到幂等性。RocketMQ 虽然不能严格保证不重复，但是正常情况下很少会出现重复发送、消费情况，只有网络异常，Consumer 启停等异常情况下会出现消息重复。

# 10、Broker 的 Buffer 满了怎么办？ #

Broker 的 Buffer 通常指的是 Broker 中一个队列的内存 Buffer 大小，这类 Buffer 通常大小有限，如果Buffer 满了以后怎么办？

CORDB Notification 规范中的处理方式：

（1）RejectNewEvents

拒绝新来的消息，向 Producer 返回 RejectNewEvents 错误码。

（2）按照特定策略丢弃已有消息。

RocketMQ 没有内存 Buffer 概念，RocketMQ 的队列都是持久化磁盘，数据定期清除。

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

# 11、回溯消费 #

回溯消费是指 Consumer 已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker 在向 Consumer 投递成功消息后，消息仍然需要保留。

RocketMQ 支持按照时间回溯消费，时间维度精确到毫秒，可以向前回溯，也可以向后回溯。

# 12、消息堆积 #

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

（1）消息堆积在内存 Buffer，一旦超过内存 Buffer，可以根据一定的丢失策略来丢弃消息。适合能容忍丢弃消息的业务，这种情况消息的堆积能力主要在于 内存 Buffer大小，而且消息堆积后，性能下降不会太大，因为内存中数据多少对于对外提供的访问能力影响有限。

（2）消息堆积到持久化存储系统中，例如DB、KV存储，文件记录形式。

当消息不能在 内存 Cache 命中时，要不可避免的访问磁盘，会产生大量读 IO、读IO的吞吐量直接决定了消息堆积后的访问能力。

评估消息堆积能力主要有以下四点：

1）消息能堆积多少条，多少字节？即消息的堆积容量。

2）消息堆积后，发消息的吞吐量大小，是否会受堆积影响？

3）消息堆积后，正常消费的 Consumer 是否会受影响？

4）消息堆积后，访问堆积在磁盘的消息时，吞吐量有多大？

# 13、分布式事务 #

已知的几个分布式事务规范，如 XA、JTA等。其中 XA规范被各大数据库厂商广泛支持。

分布式事务涉及到两阶段提交问题，在数据存储方面必须需要 KV 存储的支持，因为第二阶段的提交回滚需要修改消息状态，一定涉及到根据 key 去查找 Message 的动作。RocketMQ 在第二阶段绕过了根据 key 去查找 Message 的问题，采用第一阶段发送 Prepared 消息时，拿到了消息的 Offset，第二阶段通过 Offset 去访问消息，并修改状态，Offset 就是数据的地址。

RocketMQ 这种实现事务方式，没有通过 KV存储做，而是通过 Offset 方式，存在一个显著缺陷，即通过 Offset更改数据，会令系统的脏页过多，需要特别关注。

# 14、定时消息 #

定时消息是指消息发到 Broker 后，不能立即被 Consumer 消费，要到特定的时间点或者等待特定的时间后才能被消费。

如果要支持任意的时间精度，在 Broker 层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。

RocketMQ 支持定时消息，但是不支持任意时间精度，支持特定的 level，例如定时 5s、10s等。

# 15、消息重试 #

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

1. 由于消息本身的原因，例如反序列化失败，消息数据本身无法处理。

2. 由于依赖的下游应用服务不可用，例如db连接不可用等。