消息队列设计

曾经终败给现在 2022-10-24 15:21 49阅读 0赞

# 消息队列设计 #

## 前言 ##

最近被问到这样的问题，如何让你设计一套消息队列，你会从哪些角度去设计。于是我回顾了自己对目前两款主流的消息队列`RabbitMQ`与`Kafka`的原理，并收集了一些大佬们对此问题的解答，于是给出了自己的一些理解与认识。

**消息队列**可参考我的博客[溪源的Java笔记—消息队列][Java]

## 正文 ##

## 消息队列 ##

**消息队列要实现的主要功能**

*  **解耦**：基于消息的模型，关心的是“通知”，而非“处理”，相对而言更关心结果而不是过程，通过消息队列可以减少系统与系统的耦合性，换句话来说一个系统的低效率不会拖累其他系统。
 *  **最终一致性**：由于强一致性的成本过高，实际上我们在设计消息队列时会选择最终一致的方案：主要使用“记录”和“补偿”的方式。
 *  **广播**：采用发布—订阅的方式，生产者只需要发布消息即可，不需要去维护、关心谁会消费这些消息。
 *  **错峰与流控**：当生产者的发布速率与消费者消费速率失衡时，要使用负载均衡、限流等手段来进行流量削峰。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]

## 队列的本质 ##

*  **一次RPC变两次RPC**：从直接一次`RPC`调用接口的方式变成两次`RPC`:发布消息与消费消息。
 *  **内容存储**：借助`broker`存储内容，避免消息丢失。
 *  **选择合适的时机进行投递**：合适的时机可以理解为错峰平稳地去投递消息，避免消息堆积。

## 队列设计重点 ##

**RPC通信协议**

可以选择`Thrift`、`Dubbo`等`RPC`框架，也可以利用 `Memchached`或者`Redis`协议重新写一套`RPC`框架，但实际推荐使用前者。  
`RabbitMQ`使用的`AMQP`协议，`Kafka`采用的是一套自行设计的基于`TCP`层的协议。

**存储选型**

从速度来看，`文件系统>分布式KV（持久化）>分布式文件系统>数据库`，而可靠性却截然相反， 如果要求单`broker` 5位数以上的`QPS`性能，基于文件的存储是比较好的解决方案。整体上可以采用`数据文件+索引文件`的方式处理。

**Rabbitmq的存储**

所有队列中的消息都以`append`的方式写到一个文件中，当这个文件的大小超过指定的限制大小后，关闭这个文件再创建一个新的文件供消息的写入。文件名（`*.rdq`）从0开始然后依次累加。当某个消息被删除时，并不立即从文件中删除相关信息，而是做一些记录，当垃圾数据达到一定比例时，启动垃圾回收处理，将逻辑相邻的文件中的数据合并到一个文件中。

**Kafka的存储**

每一个`partion`(文件夹)相当于一个巨型文件被平均分配到多个大小相等`segment`(段)数据文件里。

由2大部分组成。分别为`index file`和`data file`，此2个文件一一相应，成对出现，后缀”`.index`”和“`.log`”分别表示为`segment`索引文件、数据文件。

**消费关系处理**

消费关系通常来说有两种：

*  **单播**：点对点，如队列消息
 *  **广播**：一对多，如主题消息

广播关系的维护，一般由于消息队列本身都是集群，所以都维护在公共存储上，如`config server`、`zookeeper`等。

`RabbitMQ`和`Kafka`都支持队列消息与主题消息。

**实现事务**

事务的`ACID`特性包括原子性、一致性、隔离性和持久性，其中最重要的时一致性，实现的方法主要有：

1.  两阶段提交（两阶段提交协议成本太高，并且对于仲裁`down`机或者单点故障对业务影响很大）
2.  本地事务，本地落地，补偿发送。(基于事务消息即：发送一条消息到消息中间件，然后执行本地事务，当本地事务成功后再发送提交确认到消息中间件，然后这条消息才能被其他业务消费者所能感知)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 1]

**RabbitMQ的事务**

`RabbitMQ`支持两种实现事务的方式：

*  **transaction模式**： `txSelect()`, `txCommit()`以及`txRollback()`,`txSelect`用于将当前`channel`设置成`transaction`模式，`txCommit`用于提交事务，`txRollback`用于回滚事务
 *  **confirm模式**：相对`transaction`模式具有更高的消息吞吐量，它通过生产者确认与消费者确认来保证事务的原子性。

**Kafka的事务**

`Kafka`采用第二种方法实现事务的，使用 `TransactionalID`（事务的唯一标识符） 来关联进行中的事务，通过 事务协调者 (`Transaction Coordinator`)来控制事务的流程。

**防丢/防重**

消息队列的高可用，只要保证`broker`接受消息和确认消息的接口是幂等的，并且`consumer`的几台机器处理消息是幂等的，这样就把消息队列的可用性，转交给`RPC`框架来处理了。 那么怎么保证幂等呢？最简单的方式莫过于共享存储。`broker`多机器共享一个`DB`或者一个`分布式文件/kv`系统，则处理消息自然是幂等的。

防丢和防重是两个相互权衡的问题，重复投递可以很好解决丢失的问题，这样就得考虑消息的幂等性设计。

**RabbitMQ防丢的机制**

*  **生产阶段**：失败回调机制、发布者确认、消息持久化
 *  **存储阶段**：备用交换器、死信交换器、事务、高可用队列、基于事务的高可用队列、消息持久化
 *  **消费阶段**：消费者确认、消息持久化

**Kafka防丢的机制**

*  生产者确认机制
 *  生产者失败回调机制
 *  失败重试机制
 *  消费者确认机制
 *  副本机制
 *  限定`Broker`选取`Leader`机制

**消息幂等性设计方案**

*  基于数据库的主键索引/唯一索引的来实现
 *  基于`Redis`来实现，使用`set`操作具有天然的幂等性
 *  通过先查一次数据，来判断是新增操作还是更新操作
 *  通过向数据库前置一个布隆过滤器来判断数据是新数据还是旧数据，再使用主键索引来实现
 *  `Kafka`可以通过`ProducerID`和`SequenceNumber`确保消息的幂等性
 *  使用状态机来保证幂等性，订单状态可以有初始化、订购中、订购失败、订购成功，从而限制重复订购
 *  对于前端的订单采用`token`幂等性校验，防止重复点击或者网络原因导致重复提交

**异步/批量与性能**

**异步**： 解放了线程和I/O，对I/O可以使用I/O多路复用的技术，减少了建立I/O通道的性能损耗，通常来说可以使用多线程、`NIO`实现异步。

`RabbitMQ`使用`Netty`来实现异步的，`Kafka`使用了`AIO`中的`Future`方式来实现异步的。

**批量**：批量的去处理消息，能够减少网络传输的次数。

`RabbitMQ`支持批量发送与批量消费，`Kafka`可以通过设置 `batch.size` 设置批量提交的数据大小，默认是16k,当积压的消息达到这个值的时候就会统一发送（发往同一分区的消息）

`Kafka`还支持`MMAP`技术（内存映射文件）、`DMA`技术（直接内存访问）、顺序读写磁盘等方式提升性能。

## push模式 or pull模式 ##

**消息队列的两种模式：**

*  **pull模式**：消费者主动从消息中间件拉取消息，实时性比较差，但是服务器不用去关心`consumer`的状态，能够保护`cunsumer`稳定性。
 *  **push模式**：消息中间件主动将消息推送给消费者，这种模式实时性比较好，但是容易消息堆积。
 *  **主流的消息队列原则**： `producer`将消息推送到`broker`，`consumer`从`broker`拉取消息。

`RabbitMQ`既支持`pull`模式也支持`push`模式（默认），`Kafka`只支持`pull`模式。

**慢消费**

`push`模式有以下方式的导致消息堆积情况出现：

*  如果消费者的速度比发送者的速度慢很多，就会出现消息在`broker`中堆积；
 *  `broker`给`consumer`推送一堆`consumer`无法处理的消息，`consumer`不是`reject`就是`error`，然后来回踢皮球。

`pull`模式可以避免消息堆积问题：

*  `consumer`可以按需消费，不用担心自己处理不了的消息来骚扰自己
 *  `broker`堆积消息无需记录每一个要发送消息的状态，只需要维护所有消息的队列和偏移量。

**消息延迟与忙等**

`pull`模式最大的短板是消费方无法准确地决定何时去拉取最新的消息。

两种解决方案：

*  `pull`间隔延迟一般要采用 数级增长等待。比如开始等5ms，然后10ms，然后20ms，然后40ms……直到有消息到来，然后再回到5ms。
 *  如果尝试拉取失败，不是直接`return`,而是把连接挂在那里`wait`,服务端如果有新的消息到来，把连接`notify`起来

**顺序消费**

`RabbitMQ`是采用拆分`queue`来实现消息有序性的：  
拆分多个`queue`，每个`queue`对应一个`consumer`，然后这个`consumer`内部用内存队列做排队，然后分发给底部不同`worker`处理：

*  在进行拆分`queue`时，通过对唯一标识进行`hash`运算保证同一个业务数据会进入同一个`queue`。
 *  一个`queue`对应一个消费者， 消费者内部用内存队列做排队，然后分发给底层不同的 `worker` 来处理。

*  `Kafka`是无法保证全局的消息顺序性的，只能保证主题的某个分区的消息顺序性
 *  通过指定`key`的方式（比如订单号），具有相同`key`的消息会分发到同一个`partition`,同样一个业务相关的消息就会进入同一个`partition`，`partition`会内部对其进行排序保证消息的局部有序。
 *  一个`partition`(分区)对应一个`consumer`，内部单个消费者进行消费，但是并意味一个消费者是单线程。
 *  消费者端创建多个内存队列，具有相同`key`的数据都路由到同一个内存队列；然后每个线程分别消费一个内存队列即可，这样就能保证顺序性。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 4]

[Java]: https://xiyuan.blog.csdn.net/article/details/110904482
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/ada257b340ec4a4e849b859169b7e307.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221024/ac05b34f60cb4daeb0d41c82213d6633.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221024/599262d408c545aa8c5abea027dd7242.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221024/fddb74be40324eec8e5d85796e5af5c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221024/766ac6f255fa41ff9061165dd1084fe5.png