记一次加锁导致ECS服务器CPU飙高的处理

V2AS问路

记一次加锁导致ECS服务器CPU飙高的处理

阅读原文时间：2023年08月20日阅读：5

火线告警，CPU飚了

如果你很难定位线上的问题，快速回退是一个好办法。

在多年的职业历练中，我养成了一个习惯——每次执行完发版任务的第二天，都会积极关注公司相关业务群的动向，并尽可能早的到公司。

这一天，和往常一样，我在早高峰的路上奋力前行，突然群里闪现出一条业务方发出的消息。

随即便是更多的业务对接群开始炸锅。

前段时间因为数据库性能问题，已经出现了几次线上宕机的情况，被用户吐槽。(为啥出现性能问题，此处省略若干字，后续有机会再娓娓道来)。

所以，每次今天再次遇到这样的问题，我们总是显得很被动。

我和业务团队的同事一边安抚用户的情绪，一边快马加鞭奔赴公司。

火速赶到公司之后,查看了报警日志，发现部署该业务接口的ecs CPU飙高了…

当机立断，回滚到上一版本。

大约一分钟之后，我们验证了可用性，并查看ecs和数据库各项指标，正常。

于是大家一一回复了用户群，对接群终于安静了。

猜测：分布式锁是罪魁祸

大胆假设，小心求证。

代码回滚之后一切变得正常，我们可以断定此次线上问题的一定是和昨晚的发版有关。

但是，是哪个功能或者那句代码引发了ecs cpu标高呢？

第一时间闪现在脑海里面的就是“一键已读”功能。

该功能的代码大致如下(已脱敏):

@Override
public void oneKeyRead(OneKeyReadBo bo) {
    //...

    //1. 拉取的未读的会话(群聊)
    List<Long> unReadChatIds = listUnReadChatIds(loginUser.getUserId());
    if (CollectionUtil.isEmpty(unReadChatIds)) {
        log.info("当前用户没有未读会话!");
        return;

    //2. 循环处理单个群的消息已读
    CompletableFuture.runAsync(
                    () -> {
                        processOneKeyReadChats(realUnReadChatIds, loginUser);
                    })
            .exceptionally(
                    error -> {
                        log.error("批量处理未读的群会话异常：" + error, error);
                        return null;
                    });
}

@Resource
private Executor taskExecutor;
private void processOneKeyReadChats(List<Long> realUnReadChatIds, User loginUser) {
    //循环处理单个群的消息已读
    for (Long groupChatId : realUnReadChatIds) {
        OneKeyReadMessageBo oneKeyReadMessageBo=new OneKeyReadMessageBo();
        //...省略一些代码
        oneKeyReadMessage(oneKeyReadMessageBo);
    }
}

/**
 * 单独处理一个群的消息已读
*/
private void oneKeyReadMessage(OneKeyReadMessageBo bo) {
    // 批量已读，按会话加锁
    String lockCacheKey = StrUtil.format("xxx:lock:{}:{}", bo.getUserId(), bo.getChatId());

    RLock lock = redissonClient.getLock(lockCacheKey);

    boolean success = false;
    try {
        success = lock.tryLock(10, TimeUnit.SECONDS);
    } catch (InterruptedException ignored) {
    }

    if (!success) {
        log.info(StrUtil.format("用户: {}, 消息: {}, 消息一键已读失败", bo.getUserId(), bo.getChatId()));
        throw new BizException("消息已读失败");
    }

    try {
         //1. ack 已读
         //...省略若干代码

         //2.chatmember已读
        //...省略若干代码

        //3.groupMsg已读
         //...省略若干代码
    } finally {
        lock.unlock();
    }
}

从上面的代码可以看出来，循环的最底层使用了分布式锁，且锁的时长是10s。

综上可以推断, ecs cpu爆高是底层消息处理加锁导致。

代码重构，星夜上线

重构应随时随地进行。

过去我们总是对旧项目中的“老代码”嗤之以鼻。回头看自己写过的代码，难免有点"时候诸葛亮"的意思。

在这次的版本中，为了节省时间，从项目中别处复用了处理groupMsg的代码(复制粘贴确实很爽)。

但是，忽略了那个加锁的方法在单个会话的处理是适合的，却不适合大批量的处理。

于是对代码进行重构。

主要是如下几个改进：

处理未读会话提前，批量并使用同步的方式执行，后续流程异步处理。

这样做其实是为了快速相应前端，且前端立马刷新列表，让用户能够感知到群会话的未读数已经清除。
将unReadChatIds分批处理，每次最大处理1000个。防止单次处理的未读会话过大，最终到unReadMsg上消息处理量控制在一万以内。(Mysql in 的数量进行控制)。
消息未读数处理取消锁。

大致代码如下:

 @Override
    public void oneKeyRead(OneKeyReadBo bo) {

        //1. 拉取的未读的会话(群聊)
        List<Long> unReadChatIds = listMyUnReadChatIds(loginUser.getUserId(), bo.getBeginSendTime(), bo.getEndSendTime());
        if (CollectionUtil.isEmpty(unReadChatIds)) {
            log.info("当前用户没有未读会话!");
            return;
        }

        // 同步处理clear notify
        batchClearUnreadCount(unReadChatIds, loginUser.getUserId());

        //2. ack+groupMsg已读
        CompletableFuture.runAsync(
                () -> {
                    processOneKeyReadChats(unReadChatIds, loginUser);
                })
                .exceptionally(
                        error -> {
                            log.error("批量处理未读的群会话异常：" + error, error);
                            return null;
                        });
    }

    private void processOneKeyReadChats(List<Long> unReadChatIds, User loginUser) {
        //批处理
        int total = unReadChatIds.size();
        int pageSize = 1000;

        if (total > pageSize) {
            RAMPager<Long> pager = new RAMPager(unReadChatIds, pageSize);
            System.out.println("unReadChat总页数是: " + pager.getPageCount());
            Iterator<List<Long>> iterator = pager.iterator();
            while (iterator.hasNext()) {
                List<Long> curUnReadChatIds = iterator.next();
                if (CollectionUtil.isEmpty(curUnReadChatIds)) {
                    continue;
                }
                batchReadMessage(curUnReadChatIds, loginUser);
            }
        } else {
            batchReadMessage(unReadChatIds, loginUser);
        }
    }

    /**
     * 批量处理消息已读
     */
    private void batchReadMessage(List<Long> unReadChatIds, User loginUser) {

        try {
            //1. 批量ack 已读
            //...省略若干代码

            //2. groupMsg已读
            //...省略若干代码

        } catch (Exception ex) {
            log.error(StrUtil.format("batchReadMessage 异常,error:{}", ex.getMessage()));
        }
    }

下班之后，火速上线。