卡顿、ANR、死锁线上监控

前言

性能优化这块，很多人只能说出传统的分析方式，比如 ANR 分析，只会通过查看/data/arn/下的log，分析主线程堆栈、cpu、锁等信息。但是这种方法有局限性，有些高版本设备需要root权限才能访问 /data/anr/ 目录，或者，如果只是用户反馈，我们压根没法复现的情况，就很难分析问题了。

卡顿原理与监控

卡顿原理里面需要注意的一点是：存在消息屏障的情况下，*当异步消息被处理完后，如果没有及时把消息屏障消息移除，会导致同步消息一直没有机会处理，一直阻塞在nativePollOnce *。

卡顿原理

首先，我们可以看下Looper.loop的源码：

public static void loop() {
        for (;;) {
            //1、取消息
            Message msg = queue.next(); // might block
            ...
            //2、消息处理前回调
            if (logging != null) {
                logging.println(">>>>> Dispatching to " + msg.target + " " +
                        msg.callback + ": " + msg.what);
            }
            ...
            
            //3、消息开始处理
            msg.target.dispatchMessage(msg);// 分发处理消息
            ...
            
            //4、消息处理完回调
            if (logging != null) {
                logging.println("<<<<< Finished to " + msg.target + " " + msg.callback);
            }
       }
       ...
}

这个loop方法主线程循环，可以长时间运行，从代码可以看出导致卡顿的原因可能有两个地方：

• 注释1处的取消息 queue.next() 阻塞
• 注释3处 dispatchMessage 耗时太久

看下 MessageQueue#next 的代码：

Message next() {
  ...
    for (;;) {
        //1、nextPollTimeoutMillis 不为0则阻塞
        nativePollOnce(ptr, nextPollTimeoutMillis);

        synchronized (this) {
            // Try to retrieve the next message.  Return if found.
            final long now = SystemClock.uptimeMillis();
            Message prevMsg = null;
            Message msg = mMessages;
            // 2、先判断当前第一条消息是不是同步屏障消息，
            if (msg != null && msg.target == null) {
                //3、遇到同步屏障消息，就跳过去取后面的异步消息来处理，同步消息相当于被设立了屏障
                // Stalled by a barrier.  Find the next asynchronous message in the queue.
                do {
                    prevMsg = msg;
                    msg = msg.next;
                } while (msg != null && !msg.isAsynchronous());
            }
            
            //4、正常的消息处理，判断是否有延时
            if (msg != null) {
                if (now < msg.when) {
                    //4 
                    // Next message is not ready.  Set a timeout to wake up when it is ready.
                    nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);
                } else {
                    // Got a message.
                    mBlocked = false;
                    if (prevMsg != null) {
                        prevMsg.next = msg.next;
                    } else {
                        mMessages = msg.next;
                    }
                    msg.next = null;
                    if (DEBUG) Log.v(TAG, "Returning message: " + msg);
                    msg.markInUse();
                    return msg;
                }
            } else {
                //5、如果此时是同步屏障，没有取到消息，那么下次循环就走到1那里去了，nativePollOnce为-1，会一直阻塞
                nextPollTimeoutMillis = -1;
            }
          ...
}

MessageQueue 是一个链表数据结构，它的 next 方法大致流程是这样的：

1. 首先判断它头结点(第一个消息)是否是同步屏障消息，如果是，则只处理异步msg，同步msg不处理
2. 如果是同步屏障的话，若没有获取到异步消息，就会走到注释5，设置 nextPollTimeoutMillis = -1 后，下次循环就会在注释 1 处阻塞
3. 如果获取到正常的 msg ，不管是同步还是异步，处理流程都一样，先在注释4处判断是否演示，如果延时，则会给 nextPollTimeoutMillis 赋值，下次循环到 1 处就会阻塞一段时间；如果不延时，则会return ，交给 handler 处理（确实是交给Handler处理： msg.target.dispatchMessage(msg) ，在这里面会 首先判断msg.callback 、其次是Handler的mCallback，最后才是交给 Handler 的 handleMessage 处理）

简单介绍Linux 中的IO多路复用方案

Linux上IO多路复用方案有 select、poll、epoll，它们 3 个中 epoll 的性能表现是最优秀的，能支持的并发量也最大，简单介绍：

• select 操作系统提供的函数，通过它，我们可以把一个文件描述符的数组发给操作系统，让操作系统去遍历，确定哪个文件描述符可以读写，然后让我们处理。
• poll：和select 的主要区别就是，去掉了 select 只能监听 1024 个文件描述符的限制
• epoll： 主要针对select 做了3个优化：

1、内核中保存一份文件描述符集合，无序用户每次传入，只需要告诉内核修改的部分

2、内核通过异步IO事件唤醒而不是轮询的方式找到就绪的文件描述符

3、内核仅将有IO事件的文件描述符返回给用户，用户也无需遍历整个文件描述符集合

同步屏障

Android 不允许用户调用 发送同步屏障的方法，它是hide的：

 /* @hide
 */
public int postSyncBarrier() {
    return postSyncBarrier(SystemClock.uptimeMillis());
}

private int postSyncBarrier(long when) {
  ...
}

系统一些高优先级的操作会用到同步屏障消息，例如，View在绘制的时候，最终都要调用 ViewRootImpl 的 scheduleTraversals ，会往MessageQueue 中插入同步屏障 msg，之后在unscheduleTraversals中移除 ：

void scheduleTraversals() {
    if (!mTraversalScheduled) {
        mTraversalScheduled = true;
        //插入同步屏障消息
        mTraversalBarrier = mHandler.getLooper().getQueue().postSyncBarrier();
        ...
    }
}

void unscheduleTraversals() {
    if (mTraversalScheduled) {
        mTraversalScheduled = false;
        //移除同步屏障消息
        mHandler.getLooper().getQueue().removeSyncBarrier(mTraversalBarrier);
        ...
    }
}

为了保证View的绘制过程不被主线程其他任务影响，View在绘制之前会先往MessageQueue 中插入同步屏障消息，然后注册 Vsync 信号监听，Choreographer$FrameDisplayEventReceiver就是做这事：

private final class FrameDisplayEventReceiver extends DisplayEventReceiver implements Runnable {
   
    @Override
    public void onVsync(long timestampNanos, int builtInDisplayId, int frame) {
        ...
        Message msg = Message.obtain(mHandler, this);
        //1、发送异步消息
        msg.setAsynchronous(true);
        mHandler.sendMessageAtTime(msg, timestampNanos / TimeUtils.NANOS_PER_MS);
    }
  
          @Override
    public void run() {
        ...
        // 2、doFrame优先执行
        doFrame(mTimestampNanos, mFrame);
    }
}

收到 Vsync 信号，注释1会发送异步消息，保证注释2中的doFrame 方法（View真正开始绘制的方法，会调用ViewRootImpl 的 doTraversal、performTraversals）优先执行。需要注意的是，App要谨慎使用异步msg，使用不当可能会出现主线程假死的问题，排查也会比较困难。

Handler 的dispatchMessage方法

public void dispatchMessage(Message msg) {
    if (msg.callback != null) {
        handleCallback(msg);
    } else {
        if (mCallback != null) {
            if (mCallback.handleMessage(msg)) {
                return;
            }
        }
        handleMessage(msg);
    }
}

这里印证了前面说的msg 处理顺序， msg.callback -> Handler.mCallback -> handleMessage

卡顿监控方案

卡顿监控方案一：通过Looper.loop中的日志打印监控

首先，我们可以回顾上面 Looper.loop 的源码。注释2和注释4会打印日志，中间过程3是处理msg的过程，这样两段日志之间的耗时就是msg处理的耗时。Google 为我们提供了这个接口，我们只需要 Looper.getMainLooper().setMessageLogging(printer) 设置我们自己的printer就行。需要注意的是，监听到发生卡顿之后，dispatchMessage 早已调用结束，已经出栈，此时再去获取主线程的堆栈，堆栈中是不会包含卡顿代码的！。

所以，我们需要在后台开一个线程，定时获取主线程堆栈，然后以时间点作为key，堆栈信息作为value，保存到Map中，发生卡顿的时候，只需要取出卡顿时间段内的堆栈信息即可。

不过，这种方法只适合线下使用，因其存在以下缺陷：

• logging.println 存在字符拼接，频繁调用会创建大量对象，造成内存抖动
• 获取主线程堆栈，会暂停主线程的运行

卡顿监控方案二：字节码插桩

通过 Gradle Plugin + ASM ，编译期在每个方法开始和结束的位置分别插入一行代码，统计方法耗时！，行业内的轮子有 微信的 Matrix 方案。

对于这种插桩方法，需要注意的是：

• 避免方法数暴增。在方法的入口和出口应该插入相同的函数，在编译时提前给代码中每个方法分配一个独立的ID作为参数。
• 过滤简单函数。过滤一些直接return、i++ 之类的简单函数，并支持黑名单配置，对一些调用非常频繁的函数，需要添加到黑名单来降低整个方案对性能的消耗。

微信对Matrix做了大量优化，整个包体积增大1%~2%，帧率下降 2 帧以内，对性能影响可以接收，不过依然只会在灰度包使用。

ANR监控

ANR 原理

ANR的原理可以比喻成埋炸弹和拆炸弹的过程，以Service为例，在通知AMS启动服务之前，通过Handler发送演示消息，这就是埋炸弹，若是 10s 内（前台服务是20s）没人来拆炸弹，炸弹就会爆炸。在ActivityThread中创建服务对象时，调用其 onCreate 之后，就会执行remove之前发送的消息，即拆炸弹。

常见的ANR情形如下：

• Service。 前台服务在20s内未执行完成，后台服务是前台服务的10倍，200s
• 输入事件。输入事件分发超时5s，包括按键和触摸事件
• 广播。前台广播在10s内未执行完成，后台60s
• ContentProvider。在publish过程超时10s;

AppErrors

所有的ANR，最终都会调用 AppErrors 的 appNotResponding 方法，主要包括几个流程：

1. 写入event log
2. 写入 main log
3. 生成tracesFile
4. 输出ANR logcat （控制台可以看到）
5. 尝试写入traceFile
6. 输出drapbox
7. 后台ANR，直接杀进程
8. 错误报告
9. 弹出ANR dialog

关于ANR，可以看gityuan的《彻底理解安卓应用五响应机制》

ANR分析方法：导出ANR文件

导出ANR文件，即导出/data/anr/traces.txt文件，首先查看主线程，搜索main：

ANR日志有很多信息，可以看到主线程id是1（tid=1），在等待一个锁，这个锁一直被id为22的线程持有，再来看看22号线程的堆栈：

22号线程处于Blocked状态，正在等待一个锁，这个锁被id为1的线程持有，同时这个22号线程还持有一个锁，这个锁是主线程想要的。

通过ANR日志可以分析除这个ANR是死锁导致的，并且有具体的堆栈信息。这只是一种，还有其他ANR情况，比如内存不足、CPU被抢占、系统服务没有及时响应。

如果作为线上的话，在ANR发生时，可以将这个traces.txt文件上报到服务器，只不过有些手机需要root权限才能读取 /data/anr目录

ANR监控

1、抓取系统的 traces.txt

1. 当监控线程发现主线程卡死时，主动向系统发送 SIGNAL_QUIT信号
2. 等待 /data/anr/traces.txt 文件生成
3. 文件生成后进行上报

这种方案可以参考手Q的线程死锁监控与自动化分析实践，但是，这种方案存在以下问题：

• traces.txt 里面包含所有线程信息，上传后需要人工过滤分析
• 很多高版本系统需要root权限才能读取到 /data/anr 这个目录

2、ANRWatchDog

它的主要原理：

1. 开启一个线程，死循环，循环中睡眠 5s
2. 往UI线程post一个Runnable，将_tick 赋值为 0 ，将 _reported 赋值为 false
3. 线程睡眠 5s 后检查 _tick 和 _report 的字段是否被修改
4. 如果一直没有被修改，说明主线程post的Runnable 一直没有被执行，说明主线程至少卡顿5s (只能说至少，这里存在5s内的误差)
5. 将贤臣各堆栈信息输出

但是，这种方案，其实是有缺陷的，它有个时候会捕获不到 ANR ，什么原因呢？

ANRWatchDog缺点

可以用一个图片来表示：

这种情况红色表示卡顿：

1. 假设主线程卡顿了 2s 之后，ANRWatchDog 刚好开始下一轮循环，将 _tick 赋值为5，并往主线程post一个任务，执行 _tick = 0
2. 主线程过了 3s 之后刚好不卡顿了，将 _tick 置为 0 ，
3. 等到 ANRWatchDog 睡眠 5s 之后，发现 _tick = 0 ，判断并没有发生 ANR

针对 ANRWatchDog 存在的问题，可以做一个优化。

3、ANRMoitor

针对 ANRWatchDog 的漏检测问题，设计一个 ANRMoitor ，ANRWatchDog 出现问题的主要原因是，因为线程睡眠 5s ，不知道前一秒主线程是否已经出现卡顿了，如果盖层每隔 1s 检测一次，就可以把误差降低到 1s 内。我们想让子线程间隔1s执行一次任务，可以通过 HandlerThread 来实现，代码如下：

@Volatile
var mainHandlerRunEnd = true

//子线程会间隔1s调用一次这个Runnable
private val mThreadRunnable = Runnable {
    
    blockTime++
    //1、标志位 mainHandlerRunEnd 没有被主线程修改，说明有卡顿
    if (!mainHandlerRunEnd && !isDebugger()) {
        logw(TAG, "mThreadRunnable: main thread may be block at least $blockTime s")
    }

    //2、卡顿超过5s，触发ANR流程，打印堆栈
    if (blockTime >= 5) {
        if (!mainHandlerRunEnd && !isDebugger() && !mHadReport) {
            mHadReport = true
            //5s了，主线程还没更新这个标志，ANR
            loge(TAG, "ANR->main thread may be block at least $blockTime s ")
            loge(TAG, getMainThreadStack())
            //todo 回调出去，这里可以按需把其它线程的堆栈也输出
            //todo debug环境可以开一个新进程，弹出堆栈信息
        }
    }

    //3、如果上一秒没有卡顿，那么重置标志位，然后让主线程去修改这个标志位
    if (mainHandlerRunEnd) {
        mainHandlerRunEnd = false
        mMainHandler.post {
        	mainHandlerRunEnd = true
        }
        
    }

 //子线程间隔1s调用一次mThreadRunnable
    sendDelayThreadMessage()

}

具体流程：

1. 子线程每隔 1s 执行一次 mThreadRunnable，检测标志位 mainHandlerRunEnd 是否被修改
2. 假如 mainHandlerRunEnd 被如期修改为 true，则重置 mainHandlerRunEnd 为 false，继续执行步骤 1
3. 假如 mainHandlerRunEnd 没有被修改为true，说明有卡顿，累计卡顿 5s 就触发 ANR

这种方案也能在线下应用，定位到耗时代码。最好可以结合 ProcessLifecycleOwner ，应用在前台时才开启，否则停止检测。

死锁监控

就是检测等待啥锁，锁被谁持有了

形成闭环

前面分别讲了卡顿监控、ANR监控和死锁监控，可以把它们连接起来，形成闭环：

1. 发生ANR
2. 获取主线程堆栈
3. 检测死锁
4. 上报服务器
5. 结合git，定位到最后修改代码的同学，提问题单

以上内容参考蓝师傅的博客