当前位置：首页 > news >正文

20240711每日消息队列-------------MQ消息的积压的折磨

news 2025/10/23 9:56:29

目标

解决MQ消息的积压

背景

菜馆系统-----------
系统读取消息，处理业务逻辑，持久化订单和菜品数据，然后将其显示在菜品管理客户端上。
在这里插入图片描述
最初我们的用户基数很小，上线后的一段时间内，MQ消息通信还算顺利。
随着用户规模的扩大，每个商家每天都会产生大量的订单数据，每个订单都包含多个菜品。这导致我们的菜肴管理系统的数据量显着增加。
某一天，商家投诉，称用户下单后平板上的菜品列表出现延迟。
几分钟后厨房才看到菜品。

这能行？

很明显出现这样的菜品展示延迟肯定和Kafka有关，所以我们先从排查Kafka开始。
正如预期的那样，有一个 message backlog 。
通常，消息积压的原因有：

MQ 使用者已关闭。
MQ 生产者生成消息的速率超过 MQ 消费者消费消息的速率。

我们检查了监控系统，发现我们的MQ消费服务运行正常，没有任何异常。
剩下的原因可能是MQ消费者的消息处理速度变慢了。
接下来我查看了菜品管理表，只有几十万条记录。

首先定位处理MQ日志比较慢的地方：
在代码中添加了一些日志来打印出MQ消费者中各个关键点所花费的时间。
确实有两个地方延迟有点高：
1、有一段代码在for循环中，一条一条的查询数据库。
2、有一段代码执行多条件数据查询。

解决循环查询

对于在for循环中一一查询数据库的代码，我使用参数集合将其更改为 batch query 。
原代码如下：

public List<User> queryUser(List<User> searchList) {if (CollectionUtils.isEmpty(searchList)) {return Collections.emptyList();}
List<User> result = Lists.newArrayList();searchList.forEach(user -> result.add(userMapper.getUserById(user.getId())));return result;
}

改进一下：

public List<User> queryUser(List<User> searchList) {if (CollectionUtils.isEmpty(searchList)) {return Collections.emptyList();}List<Long> ids = searchList.stream().map(User::getId).collect(Collectors.toList());return userMapper.getUserByIds(ids);
}

很简单的调整，搞一个ids集合,轻松解决挤压问题。

第二次遇到消息积压

这一次，它是零星的，只是偶尔发生，而不是大多数时候发生。
查了一下菜品管理表，现在已经有几百万条记录了
通过监控和DBA每天的慢查询邮件，我注意到了一些异常情况。

我发现有些SQL语句的WHERE条件完全相同，只是参数值不同，导致使用的索引不同。
例如， order_id=123 使用索引 a，而 order_id=124 使用索引 b。
该表查询场景众多，为了满足不同的业务需求，增加了多个复合索引。

MySQL 根据几个因素选择索引：
1、通过数据采样估计要扫描的行数。更多行可能会导致更高的 I/O 操作和更高的 CPU 使用率。
2、是否使用临时表，也会影响查询速度。
3、是否需要排序，因为它会影响查询速度。

考虑到这些因素和其他因素，MySQL 优化器会选择它认为最合适的索引。

MySQL优化器通过采样来估计要扫描的行数，这涉及到选择一些数据页进行统计估计，从而引入一些误差。

由于MVCC设计，存在多个版本的数据页。例如，删除的数据可能在其他事务中仍然可见，因此索引并未真正删除。这可能会导致统计数据不准确并影响优化器的决策。
这些因素都会导致MySQL在执行SQL语句时出现错误索引
为了解决MySQL选择错误索引的问题，我们使用 FORCE INDEX 关键字强制SQL查询使用索引a。

FORCE INDEX

force index() 方法强制使用这个索引

第三次遇到消息积压

半年后的某一天，检查监控系统，发现Kafka消息再次积压。
检查了MySQL索引，发现使用了正确的索引，但数据查询仍然很慢。
检查菜品管理表，短短六个月内就增长到了 3000 万条记录。

通常，当单个表包含太多数据时，查询和写入性能都会下降。
这次查询缓慢的原因是数据量太大。

大数据表，解决这个问题，我们需要：
1、实施数据库和表分区
2、备份历史数据

但是我们的体量和预算不支持分库分表

因此，我们决定备份历史数据。
经过与产品经理和DBA讨论，我们决定菜品管理表只保留最近30天的数据，而早于该时间的数据将移至 historical table 。
经过这样的优化，菜品管理表在30天内只积累了几百万条记录，对性能的影响很小。

第四次遇到消息积压

又又又，没错，来了
年后的一个下午，当我查看公司邮件时，发现大量关于Kafka消息积压的监控警报邮件。
下午，这个时间点很奇怪。。。

经过上面的排查都没问题。
我询问订单团队当天下午是否发布了新版本或执行了任何特定功能。因为我们的菜品管理系统是他们的下游系统，跟他们的运营有直接关系。有同事提到，半个小时前，他们做了一个作业，批量更新了几万个订单的状态。更改订单状态会自动发送 MQ 消息。
这导致他们的程序在很短的时间内生成了大量的MQ消息。
我们的 MQ 使用者无法足够快地处理这些消息，导致消息积压。
我们检查了Kafka消息积压情况，发现有几十万条消息在排队。

查看Kafka消息积压情况

https://www.cnblogs.com/lanbojini/p/17314699.html

快速提高MQ消费者的处理速度

我们考虑了两种解决方案：
1、增加分区数量。
2、使用线程池来处理消息。

然而，由于消息已经积压在现有分区中，因此添加新分区并没有多大帮助。
因此，我们决定重构代码以使用 thread pool 来处理消息。
为了开始消耗积压的消息，我们将线程池的核心线程数和最大线程数增加到 50。
此次调整后，积压的数十万条消息在20分钟左右就得到了处理。
解决这个问题后，我们保留了消息消费的线程池逻辑，将核心线程数设置为 8 ，最大线程数设置为 10 。
这使我们能够临时调整线程计数，以快速解决未来的任何消息积压问题，而不会显着影响用户。

注意：使用线程池消费MQ消息并不是通用的解决方案。它有一些缺点，例如潜在的消息排序问题以及导致服务器 CPU 使用率飙升的风险。另外，如果在多线程环境下调用第三方接口，可能会导致第三方服务过载而崩溃。

结语

MQ积压，没有完美的解决方案，只有最适合当前业务场景的解决方案。fuck everythimg

目标

背景