如何降低消息队列系统中消息的延迟?

客官°小女子只卖身不卖艺 2024-02-19 08:08 101阅读 0赞

##  ##

**目录**

前言

一、如何监控消息延迟？

使用消息队列提供的工具，通过监控消息的堆积来完成

通过生成监控消息的方式来监控消息的延迟情况

二、减少消息延迟的正确姿势

优化消费代码提升性能

增加消费者的数量

三、数据倾斜

四、零拷贝技术

总结

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

## 前言 ##

`在使用消息队列的场景中，我们需要重点关注消息的积压情况，这其实是消费性能的问题，那么你要如何提升消费性能，保证更短的消息延迟呢？在我看来，你首先需要掌握如何来监控消息的延迟，因为有了数据之后，你才可以知道目前的延迟数据是否满足要求，也可以评估优化之后的效果。然后，你要掌握使用消息队列的正确姿势，以及关注消息队列本身是如何保证消息尽快被存储和投递的。`

`接下来，我们先来看看第一点：如何监控消息延迟。`

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

## **一、如何监控消息延迟****？** ##

在我看来，监控消息的延迟有两种方式：

*  使用消息队列提供的工具，通过监控消息的堆积来完成；
 *  通过生成监控消息的方式来监控消息的延迟情况。

接下来，我带你实际了解一下。

假设在开篇的场景之下，电商系统中的消息队列已经堆积了大量的消息，那么你要想监控消息的堆积情况，首先需要从原理上了解，在消息队列中消费者的消费进度是多少，因为这样才方便计算当前的消费延迟是多少。比方说，生产者向队列中一共生产了 1000 条消息，某一个消费者消费进度是 900 条，那么这个消费者的消费延迟就是 100 条消息。

### 使用消息队列提供的工具，通过监控消息的堆积来完成 ###

在 Kafka 中，消费者的消费进度在不同的版本上是不同的。

在 Kafka0.9 之前的版本中，消费进度是存储在 ZooKeeper 中的，消费者在消费消息的时候，先要从 ZooKeeper 中获取最新的消费进度，再从这个进度的基础上消费后面的消息。在 Kafka0.9 版本之后，消费进度被迁入到 Kakfa 的一个专门的 topic叫“\_\_consumer\_offsets”里面。所以，如果你了解 kafka 的原理，你可以依据不同的版本，从不同的位置，获取到这个消费进度的信息。

当然，作为一个成熟的组件，Kafka 也提供了一些工具来获取这个消费进度的信息，帮助你实现自己的监控，这个工具主要有两个：

首先，Kafka 提供了工具叫做“kafka-consumer-groups.sh”（它在 Kafka 安装包的bin 目录下）。

为了帮助你理解，我简单地搭建了一个 Kafka 节点，并且写入和消费了一些信息，然后我来使用命令看看消息累积情况，具体的命令如下：

![112b3bdba7be4a0083d447cbce03954f.png][]

图中的前两列是队列的基本信息，包括话题名和分区名；  
第三列是当前消费者的消费进度；  
第四列是当前生产消息的总数；  
第五列就是消费消息的堆积数（也就是第四列与第三列的差值）。  
通过这个命令你可以很方便地了解消费者的消费情况

### 通过生成监控消息的方式来监控消息的延迟情况 ###

你先定义一种特殊的消息，然后启动一个监控程序，将这个消息定时地循环写入到消息队列中，消息的内容可以是生成消息的时间戳，并且也会作为队列的消费者消费数据。业务处理程序消费到这个消息时直接丢弃掉，而监控程序在消费到这个消息时，就可以和这个消息的生成时间做比较，如果时间差达到某一个阈值就可以向我们报警。

![a550e8599c924cc1a44e200fe002bcf7.png][]

## 二、减少消息延迟的正确姿势 ##

###  ###

想要减少消息的处理延迟，我们需要在消费端和消息队列两个层面来完成。在消费端，我们的目标是提升消费者的消息处理能力，你能做的是：

优化消费代码提升性能；  
增加消费者的数量（这个方式比较简单）。

### 优化消费代码提升性能 ###

代码有bug导致消费者消费一直不成功，导致消息大量积压，当消息消费不成功的时候，可以新增一个topic专门用来处理这种消费不成功的消息，作用类似与死信队列，先将积压的消息处理掉，在慢慢分析死信队列中消息消费不成功的原因。

### 增加消费者的数量 ###

不过，第二种方式会受限于消息队列的实现。比如说，如果消息队列使用的是 Kafka 就无法通过增加消费者数量的方式，来提升消息处理能力。

因为在 Kafka 中，一个 Topic（话题）可以配置多个 Partition（分区），数据会被平均或者按照生产者指定的方式，写入到多个分区中，那么在消费的时候，Kafka 约定一个分区只能被一个消费者消费，为什么要这么设计呢？在我看来，如果有多个 consumer（消费者）可以消费一个分区的数据，那么在操作这个消费进度的时候就需要加锁，可能会对性能有一定的影响。

所以说，话题的分区数量决定了消费的并行度，增加多余的消费者也是没有用处的，那么你可以通过增加分区来提高消费者的处理能力。

![ae2a8a686e6e419093a5dbe9ecb0de6a.png][]

线上有时因为发送端发送消息速度过快，或者消费端处理消息过慢，导致broker积压大量未消费消息。这种情况可以修改消费端程序，让其将收到的消息快速转发到其他topic，然后再启动多个消费者同时消费新主题的不同分区。

那么，如何在不增加分区的前提下提升消费能力呢？

既然不能增加 consumer，那么你可以在一个 consumer 中提升处理消息的并行度，所以可以考虑使用多线程的方式来增加处理能力：你可以预先创建一个或者多个线程池，在接收到消息之后，把消息丢到线程池中来异步地处理，这样，原本串行的消费消息的流程就变成了并行的消费，可以提高消息消费的吞吐量，在并行处理的前提下，我们就可以在一次和消息队列的交互中多拉取几条数据，然后分配给多个线程来处理。

## 三、数据倾斜 ##

我们使用[Kafka][]时，某时需要消息消费是有序的，因此在生产者投递消息时，可能会指定分区，或者指定Key，此时可能会导致数据倾斜到某一个分区。

由于Kafka消费的特性，一个消费组同一个主分区，只能被一个消费者消费，当生产消息大于消费消息时，就会出现单一分区[数据倾斜][Link 1]。

根本原因，当指定分区，或者指定Key投递消息时。生产消息大于消费消息时，出现单一分区数据倾斜，此时消费端不能水平扩展。解决方案：

1、优化消费端的消费能力，如原来是200ms，优化后是50ms，当然可能无法优化;

2、**由于Kafka消息key设置的不合理，导致分区数据不均衡，**可以在Kafka producer处，给key加随机后缀，使其均衡。

3、二次消费用多个topic分摊分流消息，详细如下：

在消费端判定是哪个分区发生倾斜；  
为这个发生倾斜的分区，建多个topic分摊分流，二次投递消费，水平扩展服务器；  
topic分摊分流，根据指定的Key，弄一个算法，以确保相同的Key落在同一个分区；

## 四、零拷贝技术 ##

另外一个优化点是零拷贝技术，说是零拷贝，其实，我们不可能消灭数据的拷贝，只是尽量减少拷贝的次数。在读取消息队列的数据的时候，其实就是把磁盘中的数据通过网络发送给消费客户端，在实现上会有四次数据拷贝的步骤：  
1. 数据从磁盘拷贝到内核缓冲区；  
2. 系统调用将内核缓存区的数据拷贝到用户缓冲区；  
3. 用户缓冲区的数据被写入到 Socket 缓冲区中；  
4. 操作系统再将 Socket 缓冲区的数据拷贝到网卡的缓冲区中。

![f0c3affc282c4ca4b89df1060e23b0e1.png][]

操作系统提供了 Sendfile 函数，可以减少数据被拷贝的次数。使用了 Sendfile 之后，在内核缓冲区的数据不会被拷贝到用户缓冲区，而是直接被拷贝到 Socket 缓冲区，节省了一次拷贝的过程，提升了消息发送的性能。高级语言中对于 Sendfile 函数有封装，比如说在Java 里面的 java.nio.channels.FileChannel 类就提供了 transferTo 方法提供了 Sendfile的功能。

![364a5dae59b74e3f94a4614f4d372105.png][]

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

## 总结 ##

`提示：这里对文章进行总结：`

例如：以上就是今天要讲的内容，本文仅仅简单介绍了pandas的使用，而pandas提供了大量能使我们快速便捷地处理数据的函数和方法。

[112b3bdba7be4a0083d447cbce03954f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/19/d47ddbeb694d475499ca2c664df43658.png
[a550e8599c924cc1a44e200fe002bcf7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/19/9db94d759d7042fd844e72e7ddfc43b4.png
[ae2a8a686e6e419093a5dbe9ecb0de6a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/19/904290ad0e5342e2ab497c658d792116.png
[Kafka]: https://so.csdn.net/so/search?q=Kafka&spm=1001.2101.3001.7020
[Link 1]: https://so.csdn.net/so/search?q=%E6%95%B0%E6%8D%AE%E5%80%BE%E6%96%9C&spm=1001.2101.3001.7020
[f0c3affc282c4ca4b89df1060e23b0e1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/19/6cc7784b27ac4c10a5a5882435dbca35.png
[364a5dae59b74e3f94a4614f4d372105.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/19/c18e3a82883f472196b4c7f7a6845355.png