岸荚伺映前言我之前在一家餐饮公司待过两年每天中午和晚上用餐高峰期系统的并发量不容小觑。为了保险起见公司规定各部门都要在吃饭的时间轮流值班防止出现线上问题时能够及时处理。我当时在后厨显示系统团队该系统属于订单的下游业务。用户点完菜下单后订单系统会通过发kafka消息给我们系统系统读取消息后做业务逻辑处理持久化订单和菜品数据然后展示到划菜客户端。这样厨师就知道哪个订单要做哪些菜有些菜做好了就可以通过该系统出菜。系统自动通知服务员上菜如果服务员上完菜修改菜品上菜状态用户就知道哪些菜已经上了哪些还没有上。这个系统可以大大提高后厨到用户的效率。这一切的关键是消息中间件kafka如果它出现问题将会直接影响到后厨显示系统的用户功能使用。这篇文章跟大家一起聊聊我们当时出现过的消息积压问题希望对你会有所帮助。1 第一次消息积压刚开始我们的用户量比较少上线一段时间mq的消息通信都没啥问题。随着用户量逐步增多每个商家每天都会产生大量的订单数据每个订单都有多个菜品这样导致我们划菜系统的划菜表的数据越来越多。在某一天中午收到商家投诉说用户下单之后在平板上出现的菜品列表有延迟。厨房几分钟之后才能看到菜品。我们马上开始查原因。出现这种菜品延迟的问题必定跟kafka有关因此我们先查看kafka。果然出现了消息积压。通常情况下出现消息积压的原因有mq消费者挂了。mq生产者生产消息的速度大于mq消费者消费消息的速度。我查了一下监控发现我们的mq消费者服务在正常运行没有异常。剩下的原因可能是mq消费者消费消息的速度变慢了。接下来我查了一下划菜表目前不太多只有几十万的数据。看来需要优化mq消费者的处理逻辑了。我在代码中增加了一些日志把mq消息者中各个关键节点的耗时都打印出来了。发现有两个地方耗时比较长有个代码是一个for循环中一个个查询数据库处理数据的。有个多条件查询数据的代码。于是我做了有针对性的优化。将在for循环中一个个查询数据库的代码改成通过参数集合批量查询数据。有时候我们需要从指定的用户集合中查询出有哪些是在数据库中已经存在的。实现代码可以这样写public List queryUser(List searchList) {if (CollectionUtils.isEmpty(searchList)) {return Collections.emptyList();}List result Lists.newArrayList();searchList.forEach(user - result.add(userMapper.getUserById(user.getId())));return result;}这里如果有50个用户则需要循环50次去查询数据库。我们都知道每查询一次数据库就是一次远程调用。如果查询50次数据库就有50次远程调用这是非常耗时的操作。那么我们如何优化呢具体代码如下public List queryUser(List searchList) {if (CollectionUtils.isEmpty(searchList)) {return Collections.emptyList();}List ids searchList.stream().map(User::getId).collect(Collectors.toList());return userMapper.getUserByIds(ids);}提供一个根据用户id集合批量查询用户的接口只远程调用一次就能查询出所有的数据。多条件查询数据的地方增加了一个联合索引解决了问题。这样优化之后 mq消费者处理消息的速度提升了很多消息积压问题被解决了。2 第二次消息积压没想到过了几个月之后又开始出现消息积压的问题了。但这次是偶尔会积压大部分情况不会。这几天消息的积压时间不长对用户影响比较小没有引起商家的投诉。我查了一下划菜表的数据只有几百万。但通过一些监控和DBA每天发的慢查询邮件自己发现了异常。我发现有些sql语句执行的where条件是一模一样的只有条件后面的参数值不一样导致该sql语句走的索引不一样。比如order_id123走了索引a而order_id124走了索引b。有张表查询的场景有很多当时为了满足不同业务场景加了多个联合索引。MySQL会根据下面几个因素选择索引通过采样数据来估算需要扫描的行数如果扫描的行数多那可能io次数会更多对cpu的消耗也更大。是否会使用临时表如果使用临时表也会影响查询速度是否需要排序如果需要排序则也会影响查询速度。综合1、2、3以及其它的一些因素MySql优化器会选出它自己认为最合适的索引。MySQL优化器是通过采样来预估要扫描的行数的所谓采样就是选择一些数据页来进行统计预估这个会有一定的误差。由于MVCC会有多个版本的数据页比如删除一些数据但是这些数据由于还在其它的事务中可能会被看到索引不是真正的删除这种情况也会导致统计不准确从而影响优化器的判断。上面这两个原因导致MySQL在执行SQL语句时会选错索引。明明使用索引a的时候执行效率更高但实际情况却使用了索引b。为了解决MySQL选错索引的问题我们使用了关键字force index来强制查询sql走索引a。这样优化之后这次小范围的消息积压问题被解决了。3 第三次消息积压过了半年之后在某个晚上6点多钟。有几个商家投诉过来说划菜系统有延迟下单之后几分钟才能看到菜品。我查看了一下监控发现kafka消息又出现了积压的情况。查了一下MySQL的索引该走的索引都走了但数据查询还是有些慢。此时我再次查了一下划菜表惊奇的发现短短半年表中有3千万的数据了。通常情况下单表的数据太多无论是查询还是写入的性能都会下降。这次出现查询慢的原因是数据太多了。为了解决这个问题我们必须做分库分表将历史数据备份由于现阶段做分库分表的代价太大了我们的商户数量还没有走到这一步。因此我们当时果断选择了将历史数据做备份的方案。当时我跟产品和DBA讨论了一下划菜表只保留最近30天的数据超过几天的数据写入到历史表中。这样优化之后划菜表30天只会产生几百万的数据对性能影响不大。消息积压的问题被解决了。4 第四次消息积压通过上面这几次优化之后很长一段时间系统都没有出现消息积压的问题。但在一年之后的某一天下午又有一些商家投诉过来了。此时我查看公司邮箱发现kafka消息积压的监控报警邮件一大堆。但由于刚刚一直在开会没有看到。这次的时间点就有些特殊。一般情况下并发量大的时候是中午或者晚上的用餐高峰期而这次出现消息积压问题的时间是下午。这就有点奇怪了。刚开始查询这个问题一点头绪都没有。我问了一下订单组的同事下午有没有发版或者执行什么功能因为我们的划菜系统是他们的下游系统跟他们有直接的关系。某位同事说他们半小时之前执行了一个批量修改订单状态的job一次性修改了几万个订单的状态。而修改了订单状态会自动发送mq消息。这样导致他们的程序在极短的时间内产生了大量的mq消息。而我们的mq消费者根本无法处理这些消息所以才会产生消息积压的问题。我们当时一起查了kafka消息的积压情况发现当时积压了几十万条消息。要想快速提升mq消费者的处理速度我们当时想到了两个方案增加partion数量。使用线程池处理消息。但考虑到当时消息已经积压到几个已有的partion中了再新增partion意义不大。于是我们只能改造代码使用线程池处理消息了。为了开始消费积压的消息我们将线程池的核心线程和最大线程数量调大到了50。这两个参数是可以动态配置的。这样调整之后积压了几十万的mq消息在20分钟左右被消费完了。这次突然产生的消息积压问题被解决了。解决完这次的问题之后我们还是保留的线程池消费消息的逻辑将核心线程数调到8最大线程数调到10。当后面出现消息积压问题可以及时通过调整线程数量先临时解决问题而不会对用户造成太大的影响。注意使用线程池消费mq消息不是万能的。该方案也有一些弊端它有消息顺序的问题也可能会导致服务器的CPU使用率飙升。此外如果在多线程中调用了第三方接口可能会导致该第三方接口的压力太大而直接挂掉。总之MQ的消息积压问题不是一个简单的问题。虽说产生的根本原因是MQ生产者生产消息的速度大于MQ消费者消费消息的速度但产生的具体原因有多种。我们在实际工作中需要针对不同的业务场景做不同的优化。我们需要对MQ队列中的消息积压情况进行监控和预警至少能够及时发现问题。