消息队列入门学习

不念不忘少年蓝@ 2022-09-01 08:58 206阅读 0赞

# 为什么要使用消息队列？ #

MQ 的应用场景比较多，但是比较核心的应用场景是：**解耦、异步、削峰**。

*  解耦：将消息写入消息队列，需要消息的时候自己从消息队列中订阅，从而原系统不需要做任何修改。
 *  异步：将消息写入消息队列，非必要的业务逻辑以异步的方式运行，加快响应速度
 *  削峰：原系统慢慢的按照数据库能处理的并发量，从消息队列中慢慢拉取消息。在生产中，这个短暂的高峰期积压是允许的。

## 消息队列的应用场景 ##

### 解耦 ###

应用解耦场景一：比如订单系统，下单后需要调配送系统通知发货、调积分系统增加积分等。正常情况下，需要在下单完成后手动写代码调用这些接口，如果某个系统调用失败，则下单失败。这时这些系统是耦合在一起的，其中一个系统出故障，会导致整个下单失败。  
如果引入 MQ，下单后，把数据推送到 MQ 中，由配送系统、短信系统自己拉取数据进行消费，这时订单系统和配送系统、短信系统就解耦开了，这两个系统的故障也不会影响到下单的过程，下单后直接给用户返回下单成功。如下图 1、图 2 分别为解耦前后的情况。  
![消息队列-解耦场景][-]

解耦场景 2：还是订单系统，用户下单后，需要推送订单数据给其他部门系统（如大数据部门）做一些如统计分析类的工作。正常情况下也是需要再下单完成之后，给其他系统推送订单数据，这也会有一些问题，比如大数据部门需要增删传输的数据字段、其他部门也需要这个订单数据、某个部门突然又不需要这些数据了，这些操作都需要修改代码才能实现，就跟其他系统耦合在一起了。  
引入 MQ 后，就不在需要订单系统主动调其他系统的接口推送数据了，订单系统只需要把数据推送到 MQ 中，其他哪些部门哪些系统需要订单数据，自己编写代码去 MQ 拉取即可。如下图：  
![消息队列-解耦场景][- 1]

### 异步 ###

不知道大家有没有调用过第三方的系统，有过经验的童鞋都知道，说起第三方系统脑海里想到的一个字就是：坑！你永远不知道它啥时候就挂了，也不知道啥时候响应就会慢得跟蜗牛一样，稳定性实在是不好吐槽，五味杂陈啊。  
就好像订单系统，你的短信发送功能是第三方短信系统提供的，发货功能是第三方物流系统提供的。当接口故障的时候你需要自己编写复杂的重试逻辑，不管你是同步还是异步调用；重试失败后还要把数据持久化下来，用定时任务在未来某个时间重新推送，因为可能只是当时人家系统刚好不可用了，过个半小时一小时系统恢复了，那你还是需要重新推送的。这一整个逻辑是会复杂，还会影响性能。  
如果引入 MQ，那么我们只需把消息推送到 MQ 中，再从 MQ 中消费即可。如果接口不可用，直接给 MQ 返回消费失败，下次还可以重新拉取消息进行消费，不再需要手动编写复杂的重试代码等。如下图  
![消息队列-异步][- 2]

### 削峰 ###

多数情况下，系统的瓶颈都会在数据库，假设数据库每秒可以支撑 6000 个请求，在一些如秒杀、双十一等高峰期场景中，每秒的请求达到了 3W，那么这时数据库是扛不住这么高的并发的，而 MQ 一般抗住几万的并发没有任何问题。  
这时我们就可以引入 MQ，请求到达后先写入 MQ 中，再从 MQ 中慢慢消费消息落库，此时写入 MQ 的请求仍然是每秒 3W，但是从 MQ 消费的消息控制在每秒 6000，积压的请求可以在后面空闲期慢慢消费完成。这样，每秒 3W 的请求也可以扛下来，如下图  
![消息队列-削峰][- 3]

## 消息队列有什么缺点 ##

通常来说，应用一项技术，解决了一些问题的同时，也必然会引出其他的新的问题。只不过解决新问题会比旧问题更容易些，也就是新技术的引用是利大于弊的。引入 MQ，也同样会导致一些问题，它的缺点有以下几个：

*  系统可用性降低

我们知道，系统引入的外部依赖越多，就越容易挂掉。你看啊，本来你有 ABCD 四个系统，A 直接调用 BCD 三个系统的接口就 OK 了，但是你引入了一个 MQ，结果 MQ 挂了，搞得整套系统都不可用了。

*  系统复杂度提高

还是 ABCD 四个系统，本来是用的技术栈都蛮简单的，结果你引入了 MQ，导致经常会出现一些数据丢失、数据重复消费等奇奇怪怪的问题，导致系统的复杂度大大提高。

*  一致性问题

还是 ABCD 四个系统，本来 A 调用系统 BCD，调用 B 失败就不再调用系统 CD，直接回滚数据了。结果你引入 MQ，系统 A 把数据推送给 BCD 就直接返回成功了，但是只有 BC 消费消息成功了，系统 D 消费时写入失败了，你咋保证系统的数据一致性？  
既然消息队列会产生这些问题，我们使用 MQ 的时候就必须要解决这些问题，否则，我们实际上是在给未来的自己挖坑。下面我列出几个常见的需要解决的消息队列问题：

*  **如何保证消息队列的高可用？**
 *  **如何保证消息不被重复消费？**
 *  **如何处理消息丢失的问题？**
 *  **如何保证消息的顺序性？**
 *  **如何处理消息队列大量消息积压？**

## 消息队列的选型 ##

目前比较流行的消息队列中间件有：ActiveMQ、RabbitMQ、RocketMQ、Kafka。这里简单对比下不同的消息队列的优缺点：

*  ActiveMQ：单机吞吐量：万级；时效性：毫秒级；可用性：高；消息可靠性：小概率丢数据；功能支持：极其完备
 *  RabbitMQ：单机吞吐量：万级；时效性：微秒级；可用性：高；消息可靠性：基本不丢；功能支持：erlang 开发
 *  RocketMQ：单机吞吐量：十万级；时效性：毫秒级；可用性：非常高；消息可靠性：可配置 0 丢失；功能支持：分布式
 *  Kafka：单机吞吐量：十万级；时效性：毫秒级；可用性：非常高；消息可靠性：可配置 0 丢失；功能支持：分布式，一般配合大数据类的系统来进行实时数据计算、日志采集等场景，行业内的事实标准

通过以上的对比，ActiveMQ 现在使用得越来越少了，社区也不太活跃了；RabbitMQ 是开源的，也很稳定，社区也比较活跃，但是使用 erlang 开发，对公司来说不太可控，建议中小型公司使用；RabbitMQ 是使用 Java 开发的，社区相对比较活跃，对于公司也比较可控，毕竟能修改它的源码的大神也很多；Kafka 一般用在大数据领域的实时计算、日志采集等场景。

# 如何保证消息队列的高可用？ #

## 什么是高可用性？ ##

对于高可用性，维基百科上是这么描述的：高可用性（英语：high availability，缩写为 HA），IT 术语，指系统无中断地执行其功能的能力，代表系统的可用性程度。

## 如何保证消息队列的高可用？ ##

### RocketMQ 高可用 ###

RocketMQ 进程我们一般称为 Broker，Broker 通常是集群部署的，每个 Broker 需要注册到 NameServer 中。从这里我们可以知道高可用有两个地方需要解决，一是 Broker 挂了怎么办，二是 NameServer 挂了怎么办。  
对于 NameServer，它的高可用保障是集群化部署，各个 NameServer 之间互不通信，每个 NameServer 都有一份完整的 Broker 路由信息。当某一个 NameServer 挂掉后，对集群没有任何影响，只要还有一个 NameServer 存活，就能提供完整的服务。  
对于 Broker，它高可用保障是主从架构和多副本策略。Broker 有 Master 和 Slave 两种角色，Master 和 Salve 上的数据是一模一样的，Master Broker 收到消息后会同步给 Slave Broker。每个 Master Broker 和 Slave Broker 都会向所有 NameServer 注册。Master 宕机了，会重新选举一个 Slave 作为 Master 继续提供写服务，对读服务无影响，每个 Slave 都可以读但是不可以写；Slave 宕机了，对服务无影响。

### RabbitMQ 高可用 ###

RabbitMQ 不是分布式的，它有三种模式：单机模式、普通集群模式、镜像集群模式。其中只有镜像集群模式是能够保障高可用的，而生产环境不会使用单机模式。

*  普通集群模式：集群中所有节点都存储队列元数据，但是只有一个节点存队列内容，此节点宕机则数据会丢失
 *  镜像集群模式：集群中每个节点都存储完整的队列元数据和队列内容，是一份完整的镜像数据。某个节点宕机，对整体服务没有影响。缺点是不支持分布式，每台机器都有一份完整的数据。

### Kafka 高可用 ###

Kafka 高可用与 RocketMQ 类似，都是通过主从多备份架构实现的。每个 leader broker 可以有多个 follower broker，生产者生产和消费者消费均操作 leader broker 节点，当 leader broker 宕机后自动切换 follower 为 leader。

# 如何保证消息不被重复消费？ #

## 为什么会出现重复消费的问题？ ##

RabbitMQ、RocketMQ、Kafka 都有可能出现重复消费的问题，导致重复消费的原因可能出现在生产者，也可能出现在 MQ 或 消费者。这里说的重复消费问题是指同一个数据被执行了两次，不单单指 MQ 中一条消息被消费了两次，也可能是 MQ 中存在两条一模一样的消费。

*  生产者：生产者可能会重复推送一条数据到 MQ 中，为什么会出现这种情况呢？也许是一个 Controller 接口被重复调用了 2 次，没有做接口幂等性导致的；也可能是推送消息到 MQ 时响应比较慢，生产者的重试机制导致再次推送了一次消息。
 *  MQ：在消费者消费完一条数据响应 ack 信号消费成功时，MQ 突然挂了，导致 MQ 以为消费者还未消费该条数据，MQ 恢复后再次推送了该条消息，导致了重复消费。
 *  消费者：消费者已经消费完了一条消息，正准备但是还未给 MQ 发送 ack 信号时，此时消费者挂了，服务重启后 MQ 以为消费者还没有消费该消息，再次推送了该条消息。

## 如何保证消息队列的幂等性？ ##

消息的重复消费问题实际上涉及到消息者消费消息的幂等性问题。重复消费问题通常在消费者端解决，当然生产者端也最好简单控制下不要生产重复数据，但是一般情况下 MQ 是允许存在多条一样的数据的，只是消费端就不允许消费两条一样的数据，所以幂等性保障通常都是在消费者端实现。

那么消费者怎么解决重复消费问题呢？这个问题解决起来也比较简单，这里提供两种方法

*  状态判断法：消费者消费数据后把消费数据记录在 redis 中，下次消费时先到 redis 中查看是否存在该消息，存在则表示消息已经消费过，直接丢弃消息。
 *  业务判断法：通常数据消费后都需要插入到数据库中，使用数据库的唯一性约束防止重复消费。每次消费直接尝试插入数据，如果提示唯一性字段重复，则直接丢失消息。一般都是通过这个业务判断的方法就可以简单高效地避免消息的重复处理了。

## 总结 ##

上面描述了为什么会出现重复消费的问题，生产者、MQ、消费者都有可能导致消息的重复消费。重复消费问题通常是在消费者端解决，而我们一般默认 MQ 中是有可能存在两条一模一样的数据的，消费者要做幂等性处理。而幂等性处理最简单高效的处理是插表时根据唯一性字段判断，如订单号等。

# 如何处理消息丢失的问题？ #

## 为什么消息会丢失？ ##

跟消息重复问题类似，消息丢失也可能出现在生产者、MQ、消费者三者中。这三者导致消息丢失的原因是什么呢？

*  生产者：生产者推送消息到 MQ 中，由于网络抖动等原因消息没有推送到 MQ 中，或者消息推送到 MQ 中了但是 MQ 内部出错了，导致消息丢失。
 *  MQ：MQ 接收到消息后先把消息暂存在 OS Cache 中，消费者还没消费的时候 MQ 自己挂了，导致消息丢失。
 *  消费者：消费者消费到了这条消息，但是还没来得及处理，消费者自己挂了，但是消费者已经告诉了 MQ 自己已经消费完了，导致消息丢失。

## 如何解决消息丢失的问题 ##

不同的消息队列解决消息丢失的方法是不同的，下面分别介绍不同 MQ 是如何解决消息丢失问题的。

### RabbitMQ ###

#### 生产者导致消息丢失 ####

RabbitMQ 有两种方案可以避免消息丢失，一种是 RabbitMQ 的事务机制，一种是 Confirm 模式。这里先看一下事务机制。

RabbitMQ 客户端中 Channel 接口有这么几个函数，channel.txSelect 用来开启一个事务，channel.txCommit 用来提交事务，channel.txRollback 用来回滚事务。为了避免消息丢失，我们可以在发送消息前，先开启执行 txSelect 方法开启一个事务，接着发送消息，如果消息投递失败，执行 txRollback 回滚事务，再执行重试操作重新发送，如果消息投递成功，执行 txCommit 方法提交事务。

这个方案可以保证我们的消息一定是投递成功的，但是几乎没有人使用这种方案。因为这个方案是同步阻塞的，也就是一条消息发送后，一定要等到 MQ 回应之后执行了提交或回滚事务操作，才能继续往下执行。大致过程如下图所示：  
![消息丢失-RabbitMQ生产者][-RabbitMQ]  
RabbitMQ 还提供了另一种方法避免生产者消息丢失问题，那就是 Confirm 模式。Confirm 模式是这样子的，生产者发送消息后，不需要等待 MQ 的回应，MQ 接收成功后，会回调生产者的 ack 接口通知生产者消息投递成功了，如果 MQ 接收失败，会回调 nack 接口通知生产者消息投递失败了，生产者可以重新对这条消息进行投递。大致过程如下图所示：  
![消息丢失-RabbitMQ生产者][-RabbitMQ 1]

#### RabbitMQ 导致消息丢失 ####

RabbitMQ 自己弄丢了数据是由于持久化导致的。通常 RabbitMQ 接收到消息之后写入 OS Cache 中，就会给生产者返回接收成功的回应，这时如果 RabbitMQ 挂了，消息也就丢失了。解决方法可以结合生产者的 Confirm 模式，配置 RabbitMQ 持久化到磁盘之后，才给生产者返回 ack 信号。

#### 消费者导致消息丢失 ####

RabbitMQ 在消费者端弄丢数据，是由于 RabbitMQ 的默认自动提交 ack 导致的。解决方法就是关闭 RabbitMQ 的自动响应 ack 即可。这样消息没有处理完成，消费者挂了，RabbitMQ 会认为消息没有处理成功，会再次推送消息给消费者处理。  
![消息丢失-RabbitMQ消费者][-RabbitMQ 2]

### Kafka ###

#### 生产者导致消息丢失 ####

对于 Kafka 来说，生产者基本不会弄丢消息，因为生产者发送消息会等待 Kafka 响应成功，如果响应失败，生产者会自动不断地重试。

#### Kafka 弄丢了数据 ####

Kafka 通常会一台 leader + 两台 follower，当生产者消息刚写入 leader 成功，但是还没同步到 follower 时，leader 宕机了，此时会重新选举 leader，新的 leader 由于还未同步到这条数据，导致该条消息丢失。

解决办法是做一些配置，当有其他 follower 同步到了消息后才通知生产者消息接收成功了。配置如下：

*  给 topic 设置 `replication.factor`参数：这个值必须大于 1，要求每个 partition 必须有至少 2 个副本。
 *  在 Kafka 服务端设置 `min.insync.replicas` 参数：这个值必须大于 1，这个是要求一个 leader 至少感知到有至少一个 follower 还跟自己保持联系，没掉队，这样才能确保 leader 挂了还有一个 follower。
 *  在 producer 端设置 `acks=all` ：这个是要求每条数据，必须是**写入所有 replica 之后，才能认为是写成功了**。

按上面的配置配置后，就可以保证在 Kafka Broker 端就算 leader 故障了，进行新 leader 选举切换时，也不会丢失数据。

#### 消费者导致消息丢失 ####

Kafka 消费端弄丢数据原因跟 RabbitMQ 类似，Kafka 消费者会在接收到消息的时候，会自动提交一个 offset 给 Kafka，告诉 Kafka 消息已经处理了。处理方法也跟 RabbitMQ 类似，关闭 offset 的自动提交即可。

### RocketMQ ###

RocketMQ 导致数据丢失的原因与前面的 RabbitMQ 和 Kafka 都很类似。生产者就是因为网络抖动等原因消息投递失败，或者 RocketMQ 自身的 Master 节点故障，主备切换故障之类的，消费者则有可能是异步处理导致还未处理成功就给 RocketMQ 提交了 offset 标识消息已处理了。

在 RocketMQ 中，事务消息可以保证消息零丢失。RocketMQ 的事务消息流程大致如下图所示：  
![消息丢失-RocketMQ][-RocketMQ]

在上面的事务消息流程中，基于这三个业务流程：发送 half 消息 -> 处理其他业务 -> commit/rollback。我们来讨论下面的几种情况：

*  万一生产者发送 half 消息失败，怎么办？

可以做重试或记录消息到如文件、数据库等地方，直接给用户返回失败，本次请求失败。

*  万一生产者发送 half 消息成功，但是处理其他业务失败，又该怎么办呢？

生产者发送 rollback 请求回滚 RocketMQ 中该条消息，本次请求失败。

*  万一生产者发送 half 消息成功，但是 RocketMQ 由于某些原因如网络超时等导致没有响应，怎么处理？

由于 half 消息已发送成功，此时 RocketMQ 中已经有该条消息了，RocketMQ 会有一个补偿机制，补偿机制会回调你开发好的一个接口，询问你这条消息是要 commit 还是 rollback。

*  万一生产者发送 half 消息成功，但是请求 commit 或 rollback 的时候失败了呢？

这个问题与上面的问题一样，都是通过 RocketMQ 的补偿机制来处理。

## 总结 ##

上文分别从生产者、MQ 自身、消费者介绍了导致消息丢失的原因，消息丢失问题是一个比较常见但有必须解决的问题，通常使用消息队列的业务都是比较重要的业务，不能接受数据的丢失。

接着介绍了不同的 MQ 是如何解决消息丢失问题的。消费端导致的消息丢失都是由于数据还未处理成功确提前通知 MQ 消息已经处理成功了，禁止自动提交或异步操作即可，处理起来比较简单；生产者和 MQ 自身导致的消息丢失则比较难处理，RabbitMQ 使用了 Confirm 模式避免消息丢失；Kafka 则配置所有 follower 同步成功才给生产者响应推送消息成功；RocketMQ 则使用事务消息来保证消息的零丢失，针对不同的异常情况还提供了补偿机制进行处理。

# 如何保证消息的顺序性？ #

## 为什么出现顺序错乱？ ##

在生产中经常会有一些类似报表系统这样的系统，需要做 MySQL 的 binlog 同步。比如订单系统要同步订单表的数据到大数据部门的 MySQL 库中用于报表统计分析，通常的做法是基于 Canal 这样的中间件去监听订单数据库的 binlog，然后把这些 binlog 发送到 MQ 中，再由消费者从 MQ 中获取 binlog 落地到大数据部门的 MySQL 中。

在这个过程中，可能会有对某个订单的增删改操作，比如有三条 binlog 执行顺序是增加、修改、删除；消费者愣是换了顺序给执行成删除、修改、增加，这样能行吗？肯定是不行的。

### RabbitMQ 消息顺序错乱 ###

对于 RabbitMQ 来说，导致上面顺序错乱的原因通常是消费者是集群部署，不同的消费者消费到了同一订单的不同的消息，如消费者 A 执行了增加，消费者 B 执行了修改，消费者 C 执行了删除，但是消费者 C 执行比消费者 B 快，消费者 B 又比消费者 A 快，就会导致消费 binlog 执行到数据库的时候顺序错乱，本该顺序是增加、修改、删除，变成了删除、修改、增加。

如下图是 RabbitMQ 可能出现顺序错乱的问题示意图：  
![消息顺序错乱-RabbitMQ][-RabbitMQ 3]

### Kafka 消息顺序错乱 ###

对于 Kafka 来说，一个 topic 下同一个 partition 中的消息肯定是有序的，生产者在写的时候可以指定一个 key，通过我们会用订单号作为 key，这个 key 对应的消息都会发送到同一个 partition 中，所以消费者消费到的消息也一定是有序的。

那么为什么 Kafka 还会存在消息错乱的问题呢？问题就出在消费者身上。通常我们消费到同一个 key 的多条消息后，会使用多线程技术去并发处理来提高消息处理速度，否则一条消息的处理需要耗时几十 ms，1 秒也就只能处理几十条消息，吞吐量就太低了。而多线程并发处理的话，binlog 执行到数据库的时候就不一定还是原来的顺序了。

如下图是 Kafka 可能出现乱序现象的示意图：  
![消息顺序错乱-Kafka][-Kafka]

### RocketMQ 消息顺序错乱 ###

对于 RocketMQ 来说，每个 Topic 可以指定多个 MessageQueue，当我们写入消息的时候，会把消息均匀地分发到不同的 MessageQueue 中，比如同一个订单号的消息，增加 binlog 写入到 MessageQueue1 中，修改 binlog 写入到 MessageQueue2 中，删除 binlog 写入到 MessageQueue3 中。

但是当消费者有多台机器的时候，会组成一个 Consumer Group，Consumer Group 中的每台机器都会负责消费一部分 MessageQueue 的消息，所以可能消费者 A 消费了 MessageQueue1 的消息执行增加操作，消费者 B 消费了 MessageQueue2 的消息执行修改操作，消费者 C 消费了 MessageQueue3 的消息执行删除操作，但是此时消费 binlog 执行到数据库的时候就不一定是消费者 A 先执行了，有可能消费者 C 先执行删除操作，因为几台消费者是并行执行，是不能够保证他们之间的执行顺序的。

如下图是 RocketMQ 可能出现乱序现象的示意图：  
![消息顺序错乱-RocketMQ][-RocketMQ 1]

## 如何保证消息的顺序性？ ##

知道了为什么会出现顺序错乱之后，就要想办法保证消息的顺序性了。从前面可以知道，顺序错乱要么是由于多个消费者消费到了同一个订单号的不同消息，要么是由于同一个订单号的消息分发到了 MQ 中的不同机器中。不同的消息队列保证消息顺序性的方案也各不相同。

### RabbitMQ 保证消息的顺序性 ###

RabbitMQ 的问题是由于不同的消息都发送到了同一个 queue 中，多个消费者都消费同一个 queue 的消息。解决这个问题，我们可以给 RabbitMQ 创建多个 queue，每个消费者固定消费一个 queue 的消息，生产者发送消息的时候，同一个订单号的消息发送到同一个 queue 中，由于同一个 queue 的消息是一定会保证有序的，那么同一个订单号的消息就只会被一个消费者顺序消费，从而保证了消息的顺序性。  
如下图是 RabbitMQ 保证消息顺序性的方案：  
![消息顺序-RabbitMQ][-RabbitMQ 4]

### Kafka 保证消息的顺序性 ###

Kafka 从生产者到消费者消费消息这一整个过程其实都是可以保证有序的，导致最终乱序是由于消费者端需要使用多线程并发处理消息来提高吞吐量，比如消费者消费到了消息以后，开启 32 个线程处理消息，每个线程线程处理消息的快慢是不一致的，所以才会导致最终消息有可能不一致。

所以对于 Kafka 的消息顺序性保证，其实我们只需要保证同一个订单号的消息只被同一个线程处理的就可以了。由此我们可以在线程处理前增加个内存队列，每个线程只负责处理其中一个内存队列的消息，同一个订单号的消息发送到同一个内存队列中即可。  
如下图是 Kafka 保证消息顺序性的方案：  
![消息顺序-Kafka][-Kafka 1]

### RocketMQ 保证消息的顺序性 ###

RocketMQ 的消息乱序是由于同一个订单号的 binlog 进入了不同的 MessageQueue，进而导致一个订单的 binlog 被不同机器上的 Consumer 处理。

要解决 RocketMQ 的乱序问题，我们只需要想办法让同一个订单的 binlog 进入到同一个 MessageQueue 中就可以了。因为同一个 MessageQueue 内的消息是一定有序的，一个 MessageQueue 中的消息只能交给一个 Consumer 来进行处理，所以 Consumer 消费的时候就一定会是有序的。  
如下图是 RocketMQ 保证消息顺序性的方案：  
![消息顺序-RocketMQ][-RocketMQ 2]

## 总结 ##

上文介绍了不同的消息队列出现顺序错乱问题的原因，也分别给出了常用消息队列保证消息顺序性的解决方案。消息的顺序性其实是 MQ 中比较值得注意的一个常见问题，特别是对于同一订单存在多条消息的这种情况，不同的执行顺序可能导致完全不同的结果，顺序的错乱可能会导致业务上的很多问题，而且往往这些问题还是比较难排查的。不过也不是所有消息都需要考虑它的全局顺序性，不相关的消息就算顺序错乱对业务也是毫无影响的，需要根据具体问题来看。

# 如何处理消费者故障导致的百万消息积压？ #

我们先思考一下导致消息队列消息百万积压都是怎么造成的。首先，可能是消费端出问题了，比如宕机等情况，或者消费端消费突然变得极慢，就会导致消息不断积压；也有可能是消费端依赖的服务器挂掉了，比如依赖的 NoSQL/MySQL 挂掉了，导致消费者自己没法正常运作了，导致消息的积压。

## 怎么解决百万消息积压问题？ ##

如果积压的这些消息是允许丢失的，那么很简单，马上修改消费者代码直接丢弃消息即可，这个速度会很快，所以积压消息处理起来也非常地迅速。

但是往往来说，很多消息都是不允许直接丢弃的。所以我们还是需要快速地处理，怎么快速地处理呢？最简单高效的办法就是临时部署足够多的消费者，一起来消费这些消息。当然，在此之前，需要先恢复系统的正常服务。

比如对于 RocketMQ 来说，原本一个 Topic 只有 4 个 MessageQueue，对应 4 个消费者。很明显如果消息积压了百万条，那么 4 个消息消费是不能够快速处理掉这一批积压消息的。我们可以修改 4 台原消费者代码，不直接处理消息，而是先把消息发送到一台新的 RocketMQ 中，这台新的 RocketMQ 一个 Topic 有 20 个 MessageQueue，这时我们可以临时部署 20 个消费者一起消费这批数据，消息的消费速度提高了 5 倍，很快积压的百万消息都会被处理完毕。处理完积压的消息之后就可以下线临时部署的 20 台消费者了。  
![消息积压][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center]

参考  
https://www.infoq.cn/profile/BF112BC8BF6889/

[-]: /images/20220829/27bb5cb105fd45fb8f2f06aa46a7e64d.png
[- 1]: /images/20220829/d83cb846d844440c89c0d595a43acb03.png
[- 2]: /images/20220829/870c4dc1f6fe4175935be5c6c689bceb.png
[- 3]: /images/20220829/98dae697417f4ff8a544b0abeb463075.png
[-RabbitMQ]: /images/20220829/39700d9f134f48599af12936285e8fdf.png
[-RabbitMQ 1]: /images/20220829/a01b808fb4254c018c492a6a9a938fe8.png
[-RabbitMQ 2]: /images/20220829/a383143952154bc28c98cfa6dd1766d3.png
[-RocketMQ]: /images/20220829/3cfa7754468e4af2a2dc4d1092066fe3.png
[-RabbitMQ 3]: /images/20220829/f96bc1ce9bb84504b66cfd6fc32cf90b.png
[-Kafka]: /images/20220829/14fdf413964d46c6b5298c0d797b4304.png
[-RocketMQ 1]: /images/20220829/b49b3b9ccf424f6ba2609970d54c05f3.png
[-RabbitMQ 4]: /images/20220829/f0297fbeb693488197b6717ad42e34e7.png
[-Kafka 1]: /images/20220829/b68ccc92d62a4b8299e22279ed895676.png
[-RocketMQ 2]: /images/20220829/799a4d8c60b343d28979cb29559d49c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20220829/1d9afab38f8947488efc9fb7ae675750.png