ThreadPoolExecutor杂谈

痛定思痛。 2022-11-09 00:56 171阅读 0赞

# ThreadPoolExecutor杂谈 #

日常的业务开发中，避免不了使用多线程，通常我们不会在代码直接去new一个线程(不可控且耗资源)，而是借助于线程池。而java原生的线程池中应用最广泛的当属ThreadPoolExecutor

## 原理简述 ##

网上有大一堆分析ThreadPoolExecutor 的帖子，包括原理乃至源码，这里简单回顾其大概执行过程，首先看构造函数：

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler)

熟悉线程池的同学都知道如下的执行过程，随着任务数N的不断增加：  
1，N<corePoolSize，即创建Thread并执行  
2，corePoolSize<=N,任务被放进阻塞队列，直至队列满  
3，如果队列放满，且maxPoolSize>corePoolSize,则创建Thread并执行任务，直至maxPool满  
4，如果队列放满，maxPool中的线程也都在执行任务，这时再提交任务，线程池根据拒绝策略做出相应处理：

*  a)AbortPolicy:抛出RejectedExecutionException，默认策略，算是负责任的做法
 *  b)CallerRunsPolicy:用调用者所在的线程自己来执行被拒绝的任务，该策略对线程池使用者来说有点不可控，慎用。
 *  c)DiscardPolicy:悄悄地丢弃后来者，不讲武德
 *  d)DiscardOldPolicy:悄悄地丢弃(挤走)先进者，更不讲武德
    
    文字永远没有图片来得直观，流程图如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpcHNob3BfZmluX2Rldg_size_16_color_FFFFFF_t_70_pic_center]

简单总结：在使用默认拒绝策略的情况下,一个线程池最多能支持queue.size+maxPoolSize的任务同时提交。超过这个数，就会抛出RejectedExecutionException

## 注意事项 ##

关于线程池的使用规范，阿里java手册明确禁止使用Executors去创建线程池，主要原因：

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>资源使用/潜在风险</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>newSingleThreadExecutor</td> 
   <td>使用无界队列, OOM</td> 
  </tr> 
  <tr> 
   <td>newFixedThreadPool</td> 
   <td>使用无界队列, OOM</td> 
  </tr> 
  <tr> 
   <td>newCachedThreadPool</td> 
   <td>无限制线程数, OOM</td> 
  </tr> 
  <tr> 
   <td>newScheduledThreadPool</td> 
   <td>无限制线程数, OOM</td> 
  </tr> 
 </tbody> 
</table>

此除之外，结合一些踩坑经验,个人认为以下几点值得注意。

### 避免创建过多的线程池 ###

线程池作为系统资源，应该严格管理，尽量避免创建过多的线程池。我们经常会在同一个工程中看到多个线程池，由于种种原因(历史的或个人的)，不同的开发同学使用不同的线程池，这种情况下，应该按功能场景将类似的线程池进行合并，以提高资源利用率。

### 避免阻塞队列过小 ###

前一阵子，生产上出现一个奇怪的中断异常：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpcHNob3BfZmluX2Rldg_size_16_color_FFFFFF_t_70_pic_center 1]

这个问题有点不好查，刚开始还以为是空指针引起，后面排查才知道是线程池的阻塞队列太小，小于一次性提交(invokeAll)的任务数而导致(maxPoolSize设置为了10,队列设置为50,而一起提交了近100个任务)。奇怪的是为什么任务数超限不是报RejectedExecutionException,而是报InterruptedException呢？下面贴一下invokeAll源码

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException { 
            if (tasks == null)
                throw new NullPointerException();
            ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
            boolean done = false;
            try { 
                //循环调用execute
                for (Callable<T> t : tasks) { 
                    RunnableFuture<T> f = newTaskFor(t);
                    futures.add(f);
                    //当提交的任务数>阻塞队列大小+maxPool大小,触发RejectedExecutionException,这时done=false
                    execute(f);
                }
                for (int i = 0, size = futures.size(); i < size; i++) { 
                    Future<T> f = futures.get(i);
                    if (!f.isDone()) { 
                        try { 
                            f.get();
                        } catch (CancellationException ignore) { 
                        } catch (ExecutionException ignore) { 
                        }
                    }
                }
                done = true;
                return futures;
            } finally { 
                //触发任务取消:
                if (!done)
                    for (int i = 0, size = futures.size(); i < size; i++)
                        futures.get(i).cancel(true);
            }
        }

再进入FetureTask.cancel(),果然在FutureTask.cancel()中对执行该task的thread进行中断了，于是产生如上异常.

public boolean cancel(boolean mayInterruptIfRunning) { 
            if (!(state == NEW &&
                  UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                      mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
                return false;
            try {     // in case call to interrupt throws exception
                if (mayInterruptIfRunning) { 
                    try { 
                        Thread t = runner;
                        if (t != null)
                            t.interrupt();//任务中断
                    } finally {  // final state
                        UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                    }
                }
            } finally { 
                finishCompletion();
            }
            return true;
        }

### 异常处理 ###

通常我们会在任务中进行异常处理，但有时总有些漏网之鱼，有时运行时异常难免会产生。**对于一个Runnable，使用execute()和submit()，其异常处理的方式是不同的。**

1.  通过execute()提交的任务,其异常需要通过给线程设置UncaughtExceptionHandler实现捕捉.
2.  通过submit()和invokeAll()提交的任务，其异常被包装在Future中,需要通过Future.get()去触发和捕捉

线程池有4个方法可以执行任务:execute(),submit(),invokeAll()/invokeAny(),其中invokeAny()业务中很少用到略去，submit()和invokeAll()最终都是会调用execute()，所以分析execute()是关键，execute()主要是上面的流程图，其中有个addWorker的关键逻辑，将提交的任务包装成Worker对象，并由线程池中的线程来执行runWorker():

final void runWorker(Worker w) { 
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try { 
                while (task != null || (task = getTask()) != null) { 
                    w.lock();
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try { 
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try { 
                        	//执行任务:对于submit()/invokeAll()进来的任务，会被包装成FutureTask
                            task.run();
                        } catch (RuntimeException x) { 
                            thrown = x; throw x;//异常被catch后又随即抛出
                        } catch (Error x) { 
                            thrown = x; throw x;
                        } catch (Throwable x) { 
                            thrown = x; throw new Error(x);
                        } finally { 
                            afterExecute(task, thrown);
                        }
                    } finally { 
                        task = null;
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally { 
                processWorkerExit(w, completedAbruptly);
            }
        }

再看FutureTask.run

public void run() { 
            if (state != NEW ||
                !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                             null, Thread.currentThread()))
                return;
            try { 
                Callable<V> c = callable;
                if (c != null && state == NEW) { 
                    V result;
                    boolean ran;
                    try { 
                        result = c.call();
                        ran = true;
                    } catch (Throwable ex) { 
                        result = null;
                        ran = false;
                        setException(ex);//将异常进入outcome变量
                    }
                    if (ran)
                        set(result);//将任务结果进入outcome变量
                }
            } finally { 
                runner = null;
                int s = state;
                if (s >= INTERRUPTING)
                    handlePossibleCancellationInterrupt(s);
            }
        }

可以见看到runWorker虽然捕捉到任务运行时的异常，但随即抛出了异常。但是通过Submit入口进来的任务，会先被适配成Callable，再包装成FutureTask,而FutureTask重写了run()方法,并在run()方法对异常进行了捕捉，并将异常对象赋值给内部变量outcome,当调用get()方法时便可以抛出该异常。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpcHNob3BfZmluX2Rldg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/30bf5293f5534163a4c14c914ce600ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpcHNob3BfZmluX2Rldg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221023/173ad7bb48094a81a8d3187d347bd8d2.png