记一次Java线程池与ThreadLocal引发的血案

今天药忘吃喽~ 2024-02-19 17:16 77阅读 0赞

### 目录 ###

1.  Java线程池学习
2.  ThreadLocal学习
3.  实战bug现场show

### 1. Java线程池学习 ###

> Doug Lea大神的工厂类中的实现

#### 1.1 常用线程池介绍 ####

1.  固定线程池（Executors.newFixedThreadPool）：线程数为一个固定的值，当超出时放入链表队列，FIFO先进先出，队列满时直接抛出拒绝异常。参数：core，max线程数均一致为入参的NThread指定的数量，keepAliveTime为0L，队列为LinkedBlockingQueue链表阻塞队列，默认的AbortPolicy策略（即直接抛出拒绝异常：RejectedExecutionException）；
2.  单线程线程池（Executors.newSingleThreadExecutor）：只有一个线程执行，其余入队列；core，max均为1，其余同Fixed
3.  缓存线程池（Executors.newCachedThreadPool）：线程数会一直增长，直到Integer最大值之后，再向同步队列中添加元素，同步队列满了则抛出拒绝异常。参数：core为0，max为Integer最大值，keepAliveTime为60秒，队列为同步队列SynchronousQueue

#### 1.2 ThreadPoolExecutor学习 ####

ThreadPoolExecutor构造参数介绍

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>参数类型</th> 
   <th>参数含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>corePoolSize</td> 
   <td>int</td> 
   <td>队列未满情况下的工作线程数</td> 
  </tr> 
  <tr> 
   <td>maximumPoolSize</td> 
   <td>int</td> 
   <td>队列满之后可以突破core配置达到的最大线程数</td> 
  </tr> 
  <tr> 
   <td>keepAliveTime</td> 
   <td>long</td> 
   <td>从队列中获取任务的等待超时时间，如果获取超时会退出Worker，如果允许core线程超时，那么会移除所有Worker，如果队列有任务则保留至少一个Worker线程</td> 
  </tr> 
  <tr> 
   <td>unit</td> 
   <td>TimeUnit</td> 
   <td>超时时间单位</td> 
  </tr> 
  <tr> 
   <td>workQueue</td> 
   <td>BlockingQueue</td> 
   <td>任务数超过core配置后要放入的队列</td> 
  </tr> 
  <tr> 
   <td>threadFactory</td> 
   <td>ThreadFactory</td> 
   <td>线程工厂</td> 
  </tr> 
  <tr> 
   <td>handler</td> 
   <td>RejectedExecutionHandler</td> 
   <td>队列满之后线程数超过max配置后拒绝任务的handle实现</td> 
  </tr> 
 </tbody> 
</table>

##### 1.2.1 提交任务：execute(Runnable command) #####

1.  Worker的数量小于core配置，则添加一个Worker：addWorker，firstTask为command，core为true
2.  Worker的数量超过core配置，并且线程池处于running状态，提交任务至队列；如果提交成功，再次检查是否是running状态，如果不是则试图从队列中移除该任务并回调reject；如果依然是running状态，并且Worker数量为0，（即一瞬间提交的任务超过了core配置，并且在超过的那一刻，所有的core任务全部执行完毕），则继续添加addWorker，firstTask为null，core为false，（即将刚刚添加至队列的任务拉取出来并执行）
3.  Worker的数量超过core配置，并且队列已满，继续添加Worker，firstTask为command，core为false，（此时会突破core配置继续添加Worker数至max线程配置）；如果添加失败则回调reject拒绝任务（即超出了队列并且所有的工作线程数达到max配置均在执行）

##### 1.2.2 添加Worker：addWorker #####

1.  如果是core任务，判断是否超过core配置，超过则返回false（即添加失败），否则递增worker数量
2.  如果不是core任务，判断是否超过max配置，超过则返回false（即添加失败），否则递增worker数量
3.  将command封装为Worker添加至workers集合，执行worker
4.  如果执行失败（如果没有执行失败，Worker的递减动作在getTask或者processWorkerExit（突然完成循环才会走该逻辑）中执行），移除worker，并递减worker数量（说明此时任务已经超出max配置）
5.  尝试一次终止
6.  如果线程池为running、tidying、terminated状态之一或者（shutdown状态并且队列不为空）直接返回
7.  Worker数量不为0，终止一个空闲Worker后返回，即遍历workers，调用与Worker绑定的线程的interrupt方法，onlyOne为true则break
8.  Worker数量为0，将线程池状态改为整理中tidying，执行terminated，将线程池状态改为terminated，唤醒termination条件的所有等待

##### 1.2.3 执行Worker：run（runWorker） #####

1.  循环获取任务，如果与Worker绑定的command不为null则继续，如果为null则getTask（即从队列中获取command），如果getTask不为null则继续
2.  如果线程池状态为Stop或者整理中tidying或者terminated，则调用interrupt终止任务，否则继续
3.  执行beforeExecute
4.  执行command任务
5.  执行afterExecute
6.  task置为null，完成任务递增，继续循环
7.  循环结束处理离开动作processWorkerExit
8.  如果是突然的完成completedAbruptly=true，即没有正常的完成循环，则递减Worker数量
9.  如果是正常的完成completedAbruptly=false，将完成任务数计入completedTaskCount属性，从workers中移除Worker
10. 尝试一次终止：同上
11. 如果线程池状态为running或shutdown，如果不是突然的完成，如果允许core线程超时allowCoreThreadTimeOut则min置为0，否则使用core配置，如果min为0并且workQueue不为空（如果为空说明所有的任务都已经在Worker中执行不需要再添加Worker去拉取队列中的任务了），则min置为1，如果Worker数量大于等于min则return返回，否则添加Worker。（即线程池为running或者shutdown状态，Worker数量小于最小值，则添加一个Worker拉取队列任务执行）。**其实就是如果没有允许超时则始终保持core配置数量的Worker，如果允许超时则不需保持Worker，如果队列不为空则保持至少一个Worker即可**
12. addWorker读取队列任务，firstTask为null，core为false

##### 1.2.4 获取队列任务执行：getTask #####

1.  死循环获取任务
2.  如果线程池大于等于shutdown状态并且（大于等于stop状态或者队列为空），递减Worker数量并返回null
3.  如果允许core线程超时allowCoreThreadTimeOut为true或者Worker数量大于core配置则timed标识置为true
4.  （如果Worker数量大于max配置或者（timed与timeOut为true）），并且（Worker数量大于1或者队列为空），递减Worker数量并返回null；
5.  如果Worker数量大于max并且Worker大于1或者队列为空，Worker突破限制理应销毁，则递减后返回null，即移除当前Worker（可能会再添加新得Worker）
6.  如果允许超时（timed或者超过了core配置）并且从队列中获取任务超时（从队列中获取任务超过keepAliveTime配置），走到getTask方法说明Worker的本身绑定的firstTask已经执行完成，并且当前队列中获取任务已经超时，并且Worker数量大于1，即时队列不为空销毁当前的超时Worker也不会影响队列的消费，因为Worker数量大于1还存在另外至少一个Worker会尝试消费队列，递减后返回null，移除当前Worker（可能会再添加新得Worker）
7.  如果允许超时则按照超时配置拉取队列中的任务，否则阻塞的形式一直等待直到获取到队列中的任务
8.  如果获取的任务不为null返回，如果出现异常timeOut置为false（即继续重试），否则说明已经超时，timeOut置为true继续循环

### 2. ThreadLocal学习 ###

1.  设置数据：set
2.  获取当前线程
3.  获取线程的ThreadLocalMap属性，
4.  如果ThreadLocalMap不为空则按照当前实例为key设置value值
5.  如果ThreadLocalMap为空则创建并设置value值
6.  获取数据：get
7.  获取当前线程
8.  获取线程的ThreadLocalMap属性，
9.  如果ThreadLocalMap不为空则按照当前实例为key读取map中的value值
10. 如果ThreadLocalMap为空则设置初始化值并返回：setInitialValue，默认为null
11. ThreadLocalMap结构
12. ThreadLocalMap的entry的可以是对ThreadLocal实例的一个弱引用，value即设置的值

![1][]

### 3. 大型bug现场show ###

1.  问题描述
2.  接口服务中使用ThreadLocal记录了整个流程上下文的一个Boolean标识符，在测试期间，刚启动时业务流程正常，一段时间后发现改标识符一直处于true状态，于是展开了地网式搜查
3.  原因定位
4.  首先第一步先查看日志，发现接口业务处理线程始终使用的是同一个Worker线程，ThreadLocal每次绑定的都是同一个Worker线程，那么有可能是没有清除ThreadLocal导致的，检查代码发现确实是如此
5.  那么为什么会一直是同一个Worker线程呢？我们业务中使用的dubbo框架，于是看了我们业务代码中线程池的配置，发现使用的是SynchronousQueue的堆栈结构，由于测试期间没有并发每次都是一个一个串行的测试请求到接口，所以就出现了每次业务处理线程都与同一个Worker绑定的现象
6.  如下图
7.  step1:3个固定的Worker阻塞在getTask等待同步队列中的数据
8.  step2：当一个业务线程启动便有栈顶Worker3获取到进行处理
9.  step3：处理完成后Worker3的getTask任务再次回到栈顶
10. ![2][]
11. 解决方法
12. ThreadLocal使用的风险可以通过监控来避免此类问题的再次出现
13. 在使用完之后及时清理ThreadLocal缓存

[1]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/08482107ff1d4a9db3f508da492689e1.png
[2]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/1648f55613bb4c6fa5c562647cd54c02.png