【全网首发】一种检测线程阻塞的实现思路原创
背景介绍
在过去处理过的服务故障中,有一类比较典型的场景是业务线程被阻塞(造成阻塞的原因也是多种多样),慢慢导致业务线程池中的全部线程被阻塞,最终造成无法对外提供服务(现象则是CPU、Load、内存等指标都比较低,请求接口后响应超时或者没有响应)。
典型的案例:
【全网首发】记一次socket read导致业务线程阻塞的案例分析
【全网首发】MQ-消息堆积-业务线程阻塞案例分析
问题分析
响应时间是接口监控的黄金指标之一:假设接口接收请求的时间是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;
}
}
拦截线程
方案一
服务中几种常见业务线程:
- 基于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);
}
}
通过在应用启动参数中增加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();
上面代码总体逻辑是没有问题的,需要考虑的细节是上面代码在不同类加载器体系下的兼容问题。
Tomcat
pandora可运行jar包
idea
经验总结
从目前的三种方案来说,个人比较倾向方案二。