消息队列消息延迟解决方案，跟着做就行了

架构师修炼

发布于 2020-07-20 10:39:27

1.5K0

发布于 2020-07-20 10:39:27

文章被收录于专栏：架构师修炼

前面我们讲到了使用消息队列解决了我们电商系统的各种问题，削峰填谷、异步处理以及系统间解耦合，同时也对其重复消息问题进行了详细方案讲解（你的消息队列如何保证消息不丢失，且只被消费一次，这篇就教会你，秒杀系统每秒上万次下单请求，我们该怎么去设计）。那我们在消息队列的使用过程中还有没有需要注意的地方呢？

场景：

现在我们的电商系统中，当用户购买商品支付成功之后，我们就会异步的发送一个优惠券给用户，一般用户在自己完成订单之后，过了几分钟或者十几分钟收到优惠券，这个时间范围内其实用户都是能接受的，但如果过了好几个小时都没收到的话，用户就很有可能回来投诉说没收到优惠券。

所以，这个时候我们就需要关注我们消息队列消息延迟情况，即我们该怎么去提升消费性能，以达到更短的消息延迟？

首先，我们要对消息队列中数据进行监控，只有有了这些监控数据才能对比分析消息是否延迟以及预测会不会延迟，下面我们就来看看如何监控消息延迟。

消息延迟如何监控？

我们可以通过两种方式进行监控消息：

消息队列提供的工具，通过监控消息的堆积来完成。
通过生产监控消息来对消息延时的监控。

那下面我们就来详细看看。

消息队列工具

首先，我们得从原理理解消息延时是怎么去理解。我们上面案例中消息队列如果堆积了很多消息，我们得要知道它的消费进度是多少，这样就能很方便计算消息延迟多少。假设目前生产 1000 条消息，然后一个消费者消费 900 条，那么我们就知道了这个消费者消息延迟 100 条。

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

在 Kafka0.9 之前的版本中，消费进度是存储在 ZooKeeper 中的，消费者在消费消息的时候先要从 ZooKeeper 中获取最新的消费进度，再从这个进度的基础上消费后面的消息。

在 Kafka0.9 版本之后，消费进度被迁入到 Kakfa 的一个专门的 topic 叫“__consumer_offsets”里面。

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

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

前两列是队列的基本信息，包括topic名和分区名；
第三列是当前消费者的消费进度；
第四列是当前生产消息的总数；
第五列就是消费消息的堆积数（也就是第四列与第三列的差值）。

（2）第二个工具是JMX

Kafka 通过 JMX 暴露了消息堆积的数据，然后我们就可以通过写代码将这个堆积数据发送到我们的监控系统中去。

自己生成消息监控

首先，我们可以自定义一种特殊的消息，然后启动一个监控程序将消息定时循环的写入到消息队列中，这个消息可以是生成一个时间戳。同时这个消息是可以被消费者消费的，当业务消费到的时候就将其丢弃，而监控程序消费这个消息是，就将其生成时间和消费时间进行对比，如果超过了我们预设的一定阈值就像我们报警。

生产建议：上面两种方式都是可以监控消息延迟的，但是在实际生产中，这里推荐将他们两者进行结合来使用，比如，我们先可以在监控程序中通过JMX获取消息堆积数据，然后发送到我们的dashboard 中；同时起一个探测进程确认消息的延迟情况是怎样的。

通过上面我们都已经了解了消息延迟怎么进行监控，接下来我们再来看看怎么来提升消息的写入和消费性能，这样才能将异步消息更快的处理掉。

减少消息延时的正确姿势

我们可以通过在消费端和消息队列这两块来减少消息的延时。

1消费端

那我们在消费端该怎么处理呢？我们消费端处理的目标应该就是尽量的提升它的处理能力，可以这么做：

通过优化消费代码来提升性能。
增加消费者的数量。

不过第二种方式并不是对于所有的消费队列有效的，它是受消费队列限制的，比如Kafka 是不能通过增加消费者数量来提升消费性能的。

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

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

既然如此，那我们在不增加分区的情况下该怎么去提升消费性能呢？

我们虽然不能增加消费者，但是我们可以在消费者使用并行处理。所以我们就可以考虑使用多线程的方式来增加处理能力：

预先创建一个或者多个线程池。
拉取到消息丢到线程池中进行异步处理，将串行的消息消费变成了并行的。
不仅提高了吞吐量，还可以一次消费多拉取一些消息，分配给多个线程来处理。

2消息队列自身

如上我们学习到了怎么通过消费端来提升消息消费能力，那接下来我们要来看看消息队列自身在读取性能上该做些什么优化，其实有两个关键点：

消息存储
零拷贝技术

消息存储，应当使用本地磁盘作为存储介质。Page Cache 的存在就可以提升消息的读取速度，即使要读取磁盘中的数据，由于消息的读取是顺序的并且不需要跨网络读取数据，所以读取消息的 QPS 肯定是比普通数据库高很多很多。

零拷贝技术，说是零拷贝，其实我们不可能消灭数据的拷贝，只是尽量减少拷贝的次数。在读取消息队列的数据的时候，其实就是把磁盘中的数据通过网络发送给消费客户端，在实现上会有四次数据拷贝的步骤：

1. 数据从磁盘拷贝到内核缓冲区；2. 系统调用将内核缓存区的数据拷贝到用户缓冲区；3. 用户缓冲区的数据被写入到 Socket 缓冲区中；4. 操作系统再将 Socket 缓冲区的数据拷贝到网卡的缓冲区中。

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