Google Guava 中 EventBus 使用不当会导致什么故障？

作者：BUG侦探

2022 年 5 月 20 日
本文字数：3077 字
阅读完需：约 10 分钟

一、故障描述

4.26 号凌晨 3 点业务服务开始报警，现象表现为消费用户消息的 kafka 集群出现堆积，导致用户在聊天室房间内的状态更新延迟,最终用户在聊天室房间的 IM 长链接断开，用户在聊天室中无法发消息，无法上麦。

在问题排查过程中，了解问题流程图如下：

其中主要分为 5 个模块，分别为：用户、im 长链接器、kafka、im-session 服务、api 服务。

当用户与 im-connector 建立完长链接以后，用户进入、退出聊天室以及与聊天室的心跳事件均会通过 im-connector 写入下游的 kafka 中。

im-sesion 服务主要负责消费 kafka 中的用户事件消息，并写入用户的 session 信息中，当 kafka 到 im-session 这个消费链路出现问题以后，用户在聊天室的 session 信息错误，导致用户通过 api 拉取的 session 信息错误，最终导致了故障。

分析完整体流程以后，其中问题也就主要定位在 kafka 与消费 kafka 的 im-session 之间，im-session 消费 kafka 数据变慢，导致 im-session 更新用户房间状态延迟，kafka 消息积压，进而导致故障产生。

二、排查可能的原因

通过上述分析，我们考虑到是 kafka 消费慢导致的问题主要涉及三方面可能的原因，于是主要通过以下三个方面来排查问题：

kafka 本身问题，导致 consumer 拉取消息慢的问题
消费者问题，im-session 服务消费慢导致消息延迟的问题
网络问题，消息从 kafka 到 consumer 之间网络延迟的问题

2.1 kafka 问题？

首先我们考虑会不会是 kafka 自身问题导致，如果是 kafka 本身出现问题，比如同步数据过程导致磁盘 io 增加，拉取 kafka 数据变慢也是有可能的，或者是 topic 分区离线等也会导致积压问题，于是查看了 kafka 相关监控信息：

kafka 监控：

从上述监控中可以看出，在问题发生时 kafka 的 topic 分区是正常的处于同步状态，没有离线的 partition。

上图监控来看，故障时间点凌晨 3 点左右 controller 也没有进行切换

同时也查看了业务的 topic 的写入写出流量，在故障期间也是正常的。

在整体排查完 kafka 相关监控，基本能排出 kafka 自身问题，那么问题可能是在消费者或者网络问题上，于是继续针对二者进行排查。

2.2 网络问题？

网络问题是当时业务严重怀疑的一个原因，业务给出了服务自身监控，包括 gc、错误量、耗时等监控信息表明服务当时是正常的，消费者服务没有问题，kafka 也没有问题，那么网络问题貌似是最终问题导致的？

于是继续查看相关日志，在排查消费者日志时，其中 kafka-stat.log 日志监控发现了一条线索：

上图中监控日志的各项指标解释如下：

POLL_CNT : 从 kafka 拉取消息的批次
POLL_TIME：从 kafka 拉去消息的时间
RECV_CNT：拉取消息到本地以后，本地接收的消息数量
PROC_CNT：业务处理的消息条数
PROC_TIME：业务处理消息时间

在故障发生期间：POLL_CNT 次数减少，但是耗时同样减少，如果是网络问题，那么 POLL_TIME 应该增加才对。同时 PROC_TIME 增加，这个指标让我们考虑是不是业务问题？通过 kafka-stat.log 日志监控来看，这与网络延迟导致的结果不一致，这也侧面证明不是网络问题。

为了验证这一结果，继续针对 kafka-client 源码进行分析，查看了 KafkaConsumer 的源码，其中通过 kafka 获取数据主要是通过 poll 方法：

在 poll 方法中会多次调用 pollOnce 方法，查看 pollOnce 方法时候我们发现了问题：

pollOnce 方法的主要逻辑是，如果当前内存中已经有准备好的数据，则直接将数据返回给 poll 方法，否则通过 client 从远端 kafka broker 拉取数据。也就是说如果网络正常情况下，在故障发生以后，每次 poll 的数据应该都应该是直接从内存中获取到的，通过上述监控的日志，每个 5 秒左右才会拉一次，5 秒时间基本是能够让 sdk 拉去足够的数据。并且这也很好的解释了为什么 POLL_TIME 会减少，因为故障发生以后每次 poll 的数据是直接从内存拉取好的，直接内存操作比之前通过网络从远端拉取耗时低很多。

2.3 业务消费问题？

分析完 kafka-client 源码以及日志监控以后可以明确排除网络问题，那么问题基本确认是消费服务的问题。虽然 PROC_TIME 时间增加，但是排查消费者服务的业务逻辑发现，逻辑只是执行了两个操作：Redis 操作以及写 MQ，但是通过查看 Redis 和写 MQ 的监控居然发现二者的耗时很小，同时从 kafka-stat.log 监控来看，如果是业务处理慢，那么不会出现下图的情况：

在某一秒中可以处理 1k/s 的消息量，如果是业务逻辑问题，应该是 PROC_CNT 减少，PROC_TIME 增加才对。于是这里就产生了矛盾。

为了研究出上述问题所在，我们在隔离测试环境进行了问题复现，发现了以下问题：

业务服务从 kafka 拉取完数据以后，会将数据放入 AsyncEventBus 中执行，AsyncEventBus 中使用一个线程池执行消费逻辑：

但是查看了当时服务的线程池状态：

其中 30 个线程池中只有一个线程处于 RUNABLE 状态，那其他的线程为什么会被 BLOCKED？是不是因为线程都被 BLOCKED 了所以消费变慢？

为了排查这个问题，查看了业务逻辑，handleEventConnector 为业务逻辑的主要执行方法：

继续查看了业务逻辑的执行耗时：

trace com.immomo.moaservice.live.imsession.impl.service.ConnectorSessionService handleEventConnector

复制代码

上图可以看出，业务逻辑的耗时在 1.34ms，耗时时间很低不会导致业务积压，可以表明不是业务逻辑的耗时导致积压，那么问题就可以缩小为下图中：

kafka 没问题，网络问题也与监控相悖，业务逻辑耗时 1.3ms，那么问题应该就出现在 EventBus 上了。

三、EventBus 问题

先简单介绍下 EventBus，在业务逻辑中存在一种订阅发布的场景，跨服务的订阅发布可以使用 mq 等实现，内存中就可以使用 Guava EventBus 来实现：

单个时间 Event 可以被多个订阅方法去处理。

为了理解其中的实现逻辑，预计查看了 EventBus 实现类：

首先 EventBus 会获取所有注册的 @Subscribe 方法：

会根据 @Subscribe 的方法去创建一个 Subscriber：

创建的 Subscriber 有两类，这里有一个判断条件 isDeclaredThreadSafe(method)：

如果这个 Subscriber 申明了 @AllowConcurrentEvents 注解，则 new Subscriber(bus, listener, method)，如果没有申明了则 new SynchronizedSubscriber(bus, listener, method);

继续查看了 SynchronizedSubscriber：

SynchronizedSubscriber 在调用业务逻辑的时候，会使用 synchronized 块加锁，也就是说给了多线程但实际依旧是多线程抢占锁然后顺序执行，这样映射上面查看的 30 个线程池中只有一个线程在执行，其他的均 BLOCKED，BLOCKED 的线程都在等待锁。

继续验证次问题，诊断了 invokeSubscriberMethod 执行的耗时：

trace com.google.common.eventbus.Subscriber invokeSubscriberMethod

复制代码

从结果来看，invokeSubscriberMethod 执行时间要 200ms+，也 200ms 的开销也就是线程抢占所等待的时间，到此问题已经确认。

结论：Google guava 中的 AsyncEventBus 组件默认是同步处理，使用多线程但未设置 @AllowConcurrentEvents 会出现多线程之间的抢锁情况，导致业务消费 kafka 消息变慢产生消息积压。

四、解决方案

方案一：去除 EventBus 这一层，直接使用主线程消费 EventBus 使用场景一般在有多个订阅者情况下使用，排查业务逻辑过程中业务逻辑只有一个 Subscriber，这种情况没有费用 EventBus 必要，并且经过验证单线程消费的 qps 在 600/s，发布多个服务也能够满足业务需求。
方案二：去除 EventBus，保留多线程，event 直接放入线程池中执行方案二是对方案一的优化，直接使用多线程来执行，验证结果可以达到 1w+/s 消费水准，完全能够满足业务需求
方案三：继续使用 EventBus，在订阅 @Subscribe 处加上 @AllowConcurrentEvents 注解