深圳幻海软件技术有限公司 欢迎您！
更多

一种使用字节码增强技术检测线程阻塞的实现方式

2023-02-28

线程 new jvm
背景介绍在过去处理过的服务故障中，有一类比较典型的场景是业务线程被阻塞（造成阻塞的原因也是多种多样），慢慢导致业务线程池中的全部线程被阻塞，最终造成无法对外提供服务（现象则是CPU、Load、内存等指标都比较低，请求接口后响应超时或者没有响应）。问题分析响应时间是接口监控的黄金指标之一：假设接口接收

背景介绍

在过去处理过的服务故障中，有一类比较典型的场景是业务线程被阻塞（造成阻塞的原因也是多种多样），慢慢导致业务线程池中的全部线程被阻塞，最终造成无法对外提供服务（现象则是CPU、Load、内存等指标都比较低，请求接口后响应超时或者没有响应）。

问题分析

响应时间是接口监控的黄金指标之一：假设接口接收请求的时间是t1，接口处理完请求，响应的时间是t2，则接口响应时间是：t2-t1，将响应时间指标接入监控报警系统，当响应时间大于阈值的时候则进行报警；但是在线程被阻塞的情况下，由于接口一直没有返回，响应时间也就无法监控到。
阻塞的线程往往是业务线程，这些业务线程可能是：

  • 基于tomcat提供http服务的tomcat线程，线程名类似：http-nio-8080-exec-1
  • 基于RocketMQ的消息消费者线程，线程名类似：ConsumeMessageThread_1
  • 基于HSF Provider的线程，线程名类似：HSFBizProcessor-DEFAULT-12-thread-3
  • … …

如果我们能够在这些业务线程执行的必经路径上进行拦截，那么就能记录下线程开始执行的时间，同时启动定时器不断检查线程已执行的时间，当已执行时间大于设定的阈值则打印出线程栈进行报警；当线程正常返回则删除该线程记录，所以需要解决的主要是两个问题：

  • 如何拦截线程
  • 定时检查线程执行时间是否超过阈值

解决思路

通过问题分析，可以确定主要需要解决以下两个问题

检测阻塞线程

该模块主要做三件事:

  • 业务线程开始执行的时候，进行线程注册
  • 业务线程结束执行或抛异常的时候，删除线程注册信息
  • 定时检测注册的线程是否发生阻塞，如果发生阻塞则打印线程栈
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class BlockedThreadChecker {
    protected final static Log logger = LogFactory.getLog(BlockedThreadChecker.class);
    private static volatile BlockedThreadChecker instance;
    private final static int DELAY = 10;
    private final static int PERIOD = 1000;
    private ScheduledThreadPoolExecutor executor;
    private final Map<Thread, Task> threads = new ConcurrentHashMap<>();
    private BlockedThreadChecker(){
        logger.info("init BlockedThreadChecker... ...classloader:" + this.getClass().getClassLoader() + ",parent classloader:" + this.getClass().getClassLoader().getParent());
        int coreSize = Runtime.getRuntime().availableProcessors();
        ThreadFactory threadFactory = new ThreadFactory() {
            final AtomicInteger counter = new AtomicInteger();
            @Override
            public Thread newThread(Runnable r){
                Thread thread = new Thread(r, "BlockThreadCheckerTimer-" + counter.incrementAndGet());
                thread.setDaemon(true);
                return thread;
            }
        };

        executor = new ScheduledThreadPoolExecutor(coreSize, threadFactory);
        executor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run(){
                long now = System.currentTimeMillis();
                for(Map.Entry<Thread,Task> entry : threads.entrySet()){
                    long execStart = entry.getValue().startTime;
                    long dur = now - execStart;
                    if(dur >= entry.getValue().maxExecTime){
                        BlockedThreadException e = new BlockedThreadException(entry.getKey().getName() + " has been blocked " + dur + " ms");
                        e.setStackTrace(entry.getKey().getStackTrace());
                        logger.error(e.getMessage(),e);
                    }
                }
            }
        },DELAY,PERIOD, TimeUnit.MILLISECONDS);
    }
    public static BlockedThreadChecker getInstance(){
        if(instance != null){
            return instance;
        }

        synchronized (BlockedThreadChecker.class){
            if(instance != null){
                return instance;
            }
            instance = new BlockedThreadChecker();
        }
        return instance;
    }
    public void registerThread(Thread thread){
        registerThread(thread, new Task());
    }
    public void registerThread(Thread thread,Task task){
        threads.put(thread, task);
        logger.info("registerThread " + thread.getName());
    }
    public void unregisterThread(Thread thread){
        threads.remove(thread);
        logger.info("unregisterThread " + thread.getName());
    }
    class Task {
        long startTime = System.currentTimeMillis();
        long maxExecTime = 10000L;
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.
  • 47.
  • 48.
  • 49.
  • 50.
  • 51.
  • 52.
  • 53.
  • 54.
  • 55.
  • 56.
  • 57.
  • 58.
  • 59.
  • 60.
  • 61.
  • 62.
  • 63.
  • 64.
  • 65.
  • 66.
  • 67.
  • 68.
  • 69.
  • 70.
  • 71.
  • 72.
  • 73.
  • 74.
  • 75.
  • 76.

拦截线程

方案一

服务中几种常见业务线程：

  • 基于tomcat提供http服务的tomcat线程，通过实现自定义Filter，在Filter中完成线程的注册和取消注册操作；
  • 基于RocketMQ的消息消费者线程，根据业务需求统一实现MessageListenerConcurrently、MessageListenerOrderly等，在统一实现类中完成线程的注册和取消注册；
  • 基于HSF Provider的线程，通过实现自定义Filter，在Filter中完成线程的注册和取消注册操作。

该方案实现简单，但是对于业务侵入性比较强，侵入性强意味着业务在意识不到问题的时候，没有改变的动力。

方案二

基于jvm-sandbox实现自定义module，实现思路如下：

import com.alibaba.jvm.sandbox.api.Information;
import com.alibaba.jvm.sandbox.api.LoadCompleted;
import com.alibaba.jvm.sandbox.api.Module;
import com.alibaba.jvm.sandbox.api.listener.ext.Advice;
import com.alibaba.jvm.sandbox.api.listener.ext.AdviceListener;
import com.alibaba.jvm.sandbox.api.listener.ext.EventWatchBuilder;
import com.alibaba.jvm.sandbox.api.resource.ModuleEventWatcher;
import org.kohsuke.MetaInfServices;
import sun.misc.Unsafe;

import javax.annotation.Resource;
import java.lang.reflect.Field;
import java.util.Properties;

@MetaInfServices(Module.class)
@Information(id = "blocked-thread-module", version = "0.0.1", author = "yuji")
public class BlockedThreadModule implements Module, LoadCompleted {
    @Resource
    private ModuleEventWatcher moduleEventWatcher;
    private AdviceListener adviceListener = new AdviceListener() {
        @Override
        protected void before(Advice advice) throws Throwable {
            if (!advice.isProcessTop()) {
                return;
            }
            BlockedThreadChecker.getInstance().registerThread(Thread.currentThread());
        }
        @Override
        protected void afterReturning(Advice advice){
            if (!advice.isProcessTop()) {
                return;
            }
            BlockedThreadChecker.getInstance().unregisterThread(Thread.currentThread());
        }
        @Override
        protected void afterThrowing(Advice advice){
            if (!advice.isProcessTop()) {
                return;
            }
            BlockedThreadChecker.getInstance().unregisterThread(Thread.currentThread());
        }
    };
    @Override
    public void loadCompleted(){
        new EventWatchBuilder(moduleEventWatcher)
                .onClass("javax.servlet.http.HttpServlet")
                .onBehavior("service")
                .onWatch(adviceListener);

        new EventWatchBuilder(moduleEventWatcher)
                .onClass("com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently")
                .includeSubClasses()
                .onBehavior("consumeMessage")
                .onWatch(adviceListener);

        new EventWatchBuilder(moduleEventWatcher)
                .onClass("com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly")
                .includeSubClasses()
                .onBehavior("consumeMessage")
                .onWatch(adviceListener);

        new EventWatchBuilder(moduleEventWatcher)
                .onClass("com.taobao.hsf.remoting.provider.ReflectInvocationHandler")
                .includeSubClasses()
                .onBehavior("invoke")
                .onWatch(adviceListener);
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.
  • 47.
  • 48.
  • 49.
  • 50.
  • 51.
  • 52.
  • 53.
  • 54.
  • 55.
  • 56.
  • 57.
  • 58.
  • 59.
  • 60.
  • 61.
  • 62.
  • 63.
  • 64.
  • 65.
  • 66.
  • 67.
  • 68.

通过在应用启动参数中增加javaagent=jvm-sandbox agent的方式来使用，相比较方案一业务应用不需要改动任何代码，也不需要对已有封装的框架进行修改，缺点是jvm-sandbox需要提前部署到每个应用的机器上，会给运维带来工作量，个人认为这种方案是最稳定的。

方案三

为了避免方案二中运维工作，一种思路是以jar包的形式提供给业务方使用，业务方引入jar包就可以了
，主要面临两个问题需要解决。

如何触发jar包执行初始化逻辑

一种方式是通过spring boot starter的方式，比如
arthas-spring-boot-starter；
一种是根据spring容器初始化流程，选择某个切入点，比如通过实现ApplicationListener接口，监听spring初始化完成的ApplicationEvent来实现。

如何初始化jvm-sandbox

初始化的核心逻辑如下：

//通过ByteBuddyAgent获取Instrumentation
Instrumentation inst = ByteBuddyAgent.install();

//将相应版本的sandbox-spy.jar添加到BootstrapClassLoader搜索路径中
//这一步的操作是由于sandbox-spy中包名是以java开头的，所以只能通过BootstrapClassLoader进行加载
JarFile spyJarFile = new JarFile("/目录/sandbox-spy-version.jar");
inst.appendToBootstrapClassLoaderSearch(spyJarFile);
//构造jvm-sandbox CoreFeatureString
String sandboxCoreFeatureString = String.format(";system_module=%s;mode=%s;sandbox_home=%s;provider=%s;namespace=%s;unsafe.enable=true;",systemModule, "agent", sandboxHome, provider, NAMESPACE );
CoreConfigure coreConfigure = CoreConfigure.toConfigure(sandboxCoreFeatureString,null);
CoreLoadedClassDataSource classDataSource = new DefaultCoreLoadedClassDataSource(inst,true);
ProviderManager providerManager = new DefaultProviderManager(coreConfigure);
//核心类，用户自定义的module是在这个类中完成加载和初始化的
CoreModuleManager coreModuleManager = new DefaultCoreModuleManager(coreConfigure,inst,classDataSource,providerManager);
//初始化命名空间与SpyHandler对于关系
SpyUtils.init(NAMESPACE);
//加载各种module
coreModuleManager.reset();
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.

上面代码总体逻辑是没有问题的，需要考虑的细节是上面代码在不同类加载器体系下的兼容问题。

Tomcat

tomcat类加载器关系

pandora可运行jar包

pandora类加载器关系

idea

idea应用类加载器关系

经验总结

从目前的三种方案来说，个人比较倾向方案二。

参考资料

bytebuddy
jvm-sandbox
arthas

如何两天时间上线一款AI应用？
用自己的编程语言实现了一个网站（增强版）