1、背景概述
问题表象：线上某一服务突然频繁宕机，所有请求都响应超时
2、排查过程：
不得不说，其中排查过程跟唐僧取经似的，历经千辛万苦，最后才取得真经。
1）、登陆监控平台发现所有mysql请求延迟特别高。
dba也反馈数据库很多大事务未提交，查看线上日志发现许多报数据库连接池没有活跃连接数的错误
image.png
2、查看mysql qps，发现qps并不高，排除了因为并发高导致连接数不够用的原因。
image.png
并且，宕机的该台机器的内存和cpu使用率当时也不高
image.png
后面又基于arthas查看线程堆栈，查看线程是否死锁等等各种方式，都没发现问题。排查到这大家已经都快要奔溃了，有些同学头发都白了，还有些头都想秃了。这到底是啥子问题？
3、柳暗花明
突然一同事大牛，默默在群里发了个traceId，说这个trace的线程走到一半就没往下运行了。机灵的我们灵机一动,立马想起了“不堪回首”的往事。
此处涉及同事隐私，马赛克
会不会是系统中线程池“死锁”了？
果然发现是线程池“死锁”了
这段比较绕，大家可以多看八百遍。
代码片段一和代码片段二使用了同一线程池，并且代码片段一的线程创建并等待代码片段二线程的执行结果
当代码片段一开启的线程把线程池打满后代码片段二从线程池中获取不到线程时，会压入线程池队列等待线程池有资源后再执行。这就导致了代码片段一和代码片段二互相等待。即线程池“死锁”了。
这里只大概描述死锁过程，如需详细了解，请移步原作者博客：https://duapp.yuque.com/team_tech/scp/rlt7sh
//代码片段一
asyncServiceExecutor.execute(() -> {
for (DeliveryOrderDo deliveryOrderDo : deliveryOrderListDos) {
try {
deliveryOrderModifyService.partAllocateInv(deliveryOrderDo.getDeliveryOrderCode());
} catch (WmsException e) {
log.error("库存分配出错，错误信息:{}", e);
log.error("当前出库单号为:{}", deliveryOrderDo.getDeliveryOrderCode());
// throw new WmsException("当前出库单号为:" + deliveryOrderDo.getDeliveryOrderCode() + ",其" + e.getMessage());
}
}
});
//代码片段二
/*开启异步线程池进行库存分配*/
try {
allocatedPojos.stream().map(
item -> CompletableFuture.supplyAsync
(() -> {
// 多线程异步执行
return inventoryAllocated.process(item);
}, asyncServiceExecutor)
).collect(Collectors.toList()).
stream()
.map(CompletableFuture::join)
.collect(Collectors.toList());
} catch (Exception e) {
log.error("库存分配异常", e);
}
3、排查结果：
在后台疯狂调用该功能，结果确实复现了，线程池"死锁"的结论成立
至此，问题解决了一半，剩下一半就是如何去解决和避免此类问题再次发生了，毕竟，头发就那么多，掉着掉着就没了。……^ - ^
4、解决方案：
问题找到了，解决方案其实也简单，只需要主线程和子线程使用不同的线程池来处理相关业务就行了。把上述代码片段二的线程池改为使用另外一个线程池b去执行相关业务代码即可。
但是，如何去避免后续还发生这种线程池死锁问题？毕竟每次排查下来都需要花费大家大量的时间和精力(头发)。
来了来了，重点来了，只需要三分钟，买不了吃亏买不了上当
解决方案一
在创建线程池时指定ThreadFactory的newThread方法上加个主子线程是否是同一线程池的判断。
结果：本地跑测试用例后，发现并没有想象中的抛出异常。原来原因是线程池在创建时，其核心线程数都是项目启动时就被其他线程创建好了的，所以测试时使用到的子线程都是别人家的“孩子”，所以方案一是行不通了
@Data
@Slf4j
public class MyThreadFactory implements ThreadFactory {
private String threadNamePrefix;
private String nameSpace;
private AtomicInteger i = new AtomicInteger();
public MyThreadFactory(String threadNamePrefix){
this.threadNamePrefix = threadNamePrefix;
}
//都是伪代码
@Override
public Thread newThread(Runnable r) {
String currentThreadName = Thread.currentThread().getName();
if(currentThreadName.startsWith(getThreadNamePrefix())){
WmsException e = new WmsException(String.format("主线程不能与子线程共用同一个线程池poolName:%s",currentThreadName));
log.error("wms monitor 主子线程不能用同一个线程池",e);
//非生产环境，直接抛异常中断线程
if(!"csprd".equalsIgnoreCase(nameSpace)){
throw e;
}
}
Thread t = new Thread();
t.setName(threadNamePrefix+"_"+i.getAndIncrement());
return t;
}
}
解决方案二
为线程池生成代理类，在代理方法中去实现主子线程的判断逻辑，具体实现步骤如下：
步骤1、创建线程池代理类
重点说明：
1）WmsThreadPoolTaskExecutor类需要重写实现ThreadPoolTaskExecutor的所有方法
2）nameSpace变量用来控制当发生主子线程属于同一线程池时是否中断线程
3）taskExecutor变量是被代理的线程池对象
4）checkThreadPool方法用来校验是否是同一线程池
/**
* 线程池添加监控检测
* @Author: dwq
* @Date: 2021/8/12 3:27 下午
*/
@Slf4j
public class WmsThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
@Value("spring.cloud.nacos.config.namespace")
private String nameSpace;
//被代理的线程池
private ThreadPoolTaskExecutor taskExecutor;
public WmsThreadPoolTaskExecutor(ThreadPoolTaskExecutor taskExecutor){
this.taskExecutor=taskExecutor;
}
@Override
public void setCorePoolSize(int corePoolSize) {
taskExecutor.setCorePoolSize(corePoolSize);
}
@Override
public int getCorePoolSize() {
return taskExecutor.getCorePoolSize();
}
@Override
public void setMaxPoolSize(int maxPoolSize) {
taskExecutor.setMaxPoolSize(maxPoolSize);
}
@Override
public int getMaxPoolSize() {
return taskExecutor.getMaxPoolSize();
}
@Override
public void setKeepAliveSeconds(int keepAliveSeconds) {
taskExecutor.setKeepAliveSeconds(keepAliveSeconds);
}
@Override
public int getKeepAliveSeconds() {
return taskExecutor.getKeepAliveSeconds();
}
@Override
public void setQueueCapacity(int queueCapacity) {
taskExecutor.setQueueCapacity(queueCapacity);
}
@Override
public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
taskExecutor.setAllowCoreThreadTimeOut(allowCoreThreadTimeOut);
}
@Override
public void setTaskDecorator(TaskDecorator taskDecorator) {
taskExecutor.setTaskDecorator(taskDecorator);
}
@Override
public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
return taskExecutor.getThreadPoolExecutor();
}
@Override
public int getPoolSize() {
return taskExecutor.getPoolSize();
}
@Override
public int getActiveCount() {
return taskExecutor.getActiveCount();
}
@Override
public void execute(Runnable task) {
taskExecutor.execute(task);
}
@Override
public void execute(Runnable task, long startTimeout) {
taskExecutor.execute(task, startTimeout);
}
@Override
public Future<?> submit(Runnable task) {
//主线程子线程是否是同一线程池校验
checkThreadPool();
return taskExecutor.submit(task);
}
@Override
public <T> Future<T> submit(Callable<T> task) {
checkThreadPool();
return taskExecutor.submit(task);
}
@Override
public ListenableFuture<?> submitListenable(Runnable task) {
checkThreadPool();
return taskExecutor.submitListenable(task);
}
@Override
public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
checkThreadPool();
return taskExecutor.submitListenable(task);
}
@Override
public String getThreadNamePrefix() {
return taskExecutor.getThreadNamePrefix();
}
/**
* 主线程子线程是否是同一线程池校验
**/
private void checkThreadPool(){
String currentThreadName = Thread.currentThread().getName();
if(currentThreadName.startsWith(getThreadNamePrefix())){
WmsException e = new WmsException(String.format("主线程不能与子线程共用同一个线程池poolName:%s",currentThreadName));
log.error("wms monitor 主子线程不能用同一个线程池",e);
if(!"csprd".equalsIgnoreCase(nameSpace)){
throw e;
}
}
}
}
步骤2、创建WmsExecutorBeanPostProcessor类，实现spring的BeanPostProcessor接口
重点说明：
postProcessAfterInitialization方法在bean实例化后执行，此处判断对象是ThreadPoolTaskExecutor类型则返回new WmsThreadPoolTaskExecutor(threadPoolTaskExecutor);代理类
package com.poizon.scm.wms.common.executors;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
/**
* @Author: dwq
* @Date: 2021/8/12 3:13 下午
*/
public class WmsExecutorBeanPostProcessor implements BeanPostProcessor {
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if(bean instanceof ThreadPoolTaskExecutor){
ThreadPoolTaskExecutor threadPoolTaskExecutor = (ThreadPoolTaskExecutor) bean;
return new WmsThreadPoolTaskExecutor(threadPoolTaskExecutor);
}
return bean;
}
}
步骤3、初始化WmsExecutorBeanPostProcessor类
@Bean
public WmsExecutorBeanPostProcessor wmsExecutorBeanPostProcessor(){
return new WmsExecutorBeanPostProcessor();
}
步骤4、最后一步，初始化线程池
/**
* Admin后台复杂任务处理线程池
*
* @return
*/
@Bean("adminThreadPoolExecutor")
public ThreadPoolTaskExecutor adminThreadPoolExecutor() {
log.info("start adminThreadPoolExecutor");
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
//配置核心线程数
executor.setCorePoolSize(5);
//配置最大线程数
executor.setMaxPoolSize(10);
//配置队列大小
executor.setQueueCapacity(1000);
//配置线程池中的线程的名称前缀
executor.setThreadNamePrefix("adminThreadPoolExecutor-service-");
//执行初始化
executor.initialize();
return executor;
}
最后附上测试验证方法
@Test
public void testThread(){
Future<Integer> result = adminThreadPoolExecutor.submit(() -> {
log.info("开始一个线程,threadname:{}", Thread.currentThread().getName());
try {
Future<Integer> future1 = adminThreadPoolExecutor.submit(() -> {
log.info("开始一个子线程：{}", Thread.currentThread().getName());
return 2;
});
return future1.get();
} catch (Exception e) {
log.error("请求异常", e);
throw e;
}
});
try {
int count = result.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
这样即完成了一个线程池的监控。本地测试也符合期望。赶紧去试试吧
网友评论