Java threadpool机制深入分析

2022-10-15 09:21:20

简介

在前面的一篇文章里我对java threadpool的几种基本应用方法做了个总结。Java的线程池针对不同应用的场景，主要有固定长度类型、可变长度类型以及定时执行等几种。针对这几种类型的创建，java中有一个专门的Executors类提供了一系列的方法封装了具体的实现。这些功能和用途不一样的线程池主要依赖于ThreadPoolExecutor,ScheduledThreadPoolExecutor等几个类。如前面文章讨论所说，这些类和相关类的主要结构如下：

Java threadpool机制深入分析

这里不是对所有类的详细实现做一个分析，而是从现有线程池ThreadPoolExecutor的源代码出发，分析一个线程池应该考虑的要点。从本文整体的方向来说，主要结合前面文章中提交线程给线程池之后分为返回结果和不返回结果的方式，按照他们执行的脉络来分析当我们提交一个线程到线程池之后他们内部是如何运行的。顺便也详细理解线程池这种参考实现的内部结构。

起始点

我们从最初使用多线程的代码开始，在一些示例代码里，我们通过Executors.newFixedThreadPool()等方法创建了一个ExecutorService类型的线程池。实际上具体实现对应的是ThreadPoolExecutor等。然后我们再使用这个对象的execute或者submit方法。前面我们了解到，通常我们用execute方法执行一个线程不通过这个方法本身返回执行结果或者我们不需要利用这个方法来获取结果。而submit方法是需要得到结果的。那么他们两者一个要结果，一个不要结果的是怎么统一起来的呢？如果我们看类AbstractExecutorService的如下代码则就可以理解了：

  1. /**
  2. * @throws RejectedExecutionException {@inheritDoc}
  3. * @throws NullPointerException       {@inheritDoc}
  4. */
  5. public Future<?> submit(Runnable task) {
  6. if (task == null) throw new NullPointerException();
  7. RunnableFuture<Void> ftask = newTaskFor(task, null);
  8. execute(ftask);
  9. return ftask;
  10. }
  11. /**
  12. * @throws RejectedExecutionException {@inheritDoc}
  13. * @throws NullPointerException       {@inheritDoc}
  14. */
  15. public <T> Future<T> submit(Runnable task, T result) {
  16. if (task == null) throw new NullPointerException();
  17. RunnableFuture<T> ftask = newTaskFor(task, result);
  18. execute(ftask);
  19. return ftask;
  20. }
  21. /**
  22. * @throws RejectedExecutionException {@inheritDoc}
  23. * @throws NullPointerException       {@inheritDoc}
  24. */
  25. public <T> Future<T> submit(Callable<T> task) {
  26. if (task == null) throw new NullPointerException();
  27. RunnableFuture<T> ftask = newTaskFor(task);
  28. execute(ftask);
  29. return ftask;
  30. }

前面这30多行代码没什么特别的，主要针对不同类型的方法签名。他们可以接收Runnable, Callable类型的参数。对于需要返回结果的类型，通过专门一个变量来保存结果。总体来说相当于一个简单的包装。而具体执行的代码还是要看execute方法。这里需要注意的一点就是newTaskFor(task)方法通过一个包装类将Callable变量包装成一个线程，让它可以运行。

既然前面的两种方法都归结于同一个方法execute，那么我们就来看看它的具体实现：

  1. public void execute(Runnable command) {
  2. if (command == null)
  3. throw new NullPointerException();
  4. /*
  5. * Proceed in 3 steps:
  6. *
  7. * 1. If fewer than corePoolSize threads are running, try to
  8. * start a new thread with the given command as its first
  9. * task.  The call to addWorker atomically checks runState and
  10. * workerCount, and so prevents false alarms that would add
  11. * threads when it shouldn't, by returning false.
  12. *
  13. * 2. If a task can be successfully queued, then we still need
  14. * to double-check whether we should have added a thread
  15. * (because existing ones died since last checking) or that
  16. * the pool shut down since entry into this method. So we
  17. * recheck state and if necessary roll back the enqueuing if
  18. * stopped, or start a new thread if there are none.
  19. *
  20. * 3. If we cannot queue task, then we try to add a new
  21. * thread.  If it fails, we know we are shut down or saturated
  22. * and so reject the task.
  23. */
  24. int c = ctl.get();
  25. if (workerCountOf(c) < corePoolSize) {
  26. if (addWorker(command, true))
  27. return;
  28. c = ctl.get();
  29. }
  30. if (isRunning(c) && workQueue.offer(command)) {
  31. int recheck = ctl.get();
  32. if (! isRunning(recheck) && remove(command))
  33. reject(command);
  34. else if (workerCountOf(recheck) == 0)
  35. addWorker(null, false);
  36. }
  37. else if (!addWorker(command, false))
  38. reject(command);
  39. }

这是ThreadPoolExecutor里面的代码。前面这部分的代码看起来比较困难，而且也比较难懂。别急，我们先把代码放这里。在讨论这些详细实现的思路前，我们先看看几个要实现线程池需要考虑的点。

考虑的关键点

假定我们要实现一个线程池，那么有哪些地方是我们需要认真考虑的呢？从线程池本身的定义来看，它是将一组事先创建好的线程放在一个资源池里，当需要的时候就将该线程分配给具体的任务来执行。那么，这个池子该有多大呢？我们线程池肯定要面临多个线程资源的访问，是不是本身的结构要保证线程安全呢？还有，如果线程池创建好之后我们后续有若干任务使用了线程资源，当池里面的资源使用完之后我们该如何安排呢？是给线程池扩容，创建更多的线程资源，还是增加一个队列，让一些任务先在里面排队呢？在一些极端的情况下，比如说来的任务实在是太多了线程池处理不过来，对于这些任务该怎么处理呢？是丢弃还是通知给请求方？线程执行的时候会有碰到异常或者错误的情况，这些异常我们该怎么处理？怎么样保证这些异常的处理不会导致线程池其他任务的正常运行不出错呢？

总的来说，前面的这几个问题可以归结为一下几个方面：

1. 线程池的结构。

2. 线程池的任务分配策略。

3. 线程池的异常和错误处理。

下面，我们针对这几个问题结合源代码详细的分析一下。

源代码分析

线程数量和线程池状态

在ThreadPoolExecutor里面有一个AtomicInteger的数值，它用来表示两个信息，一个是当前线程池的状态，还有一个就是当前线程的数目。因为这两部分都是糅合到一个整型数字里头，所以他们的信息访问就比较紧凑和特殊一点：

  1. private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  2. private static final int COUNT_BITS = Integer.SIZE - 3;
  3. private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
  4. // runState is stored in the high-order bits
  5. private static final int RUNNING    = -1 << COUNT_BITS;
  6. private static final int SHUTDOWN   =  0 << COUNT_BITS;
  7. private static final int STOP       =  1 << COUNT_BITS;
  8. private static final int TIDYING    =  2 << COUNT_BITS;
  9. private static final int TERMINATED =  3 << COUNT_BITS;
  10. // Packing and unpacking ctl
  11. private static int runStateOf(int c)     { return c & ~CAPACITY; }
  12. private static int workerCountOf(int c)  { return c & CAPACITY; }
  13. private static int ctlOf(int rs, int wc) { return rs | wc; }

这部分的代码看起来有点怪异，其实很好理解。我们设定的线程池里面最多可以容纳的线程数为(2^29) -1。这也就是为什么前面用一个Integer.SIZE - 3作为位数。这样这个整数的0-28位表示的就是线程的数目。而高位的部分，29-31位的地方则表示线程池的状态。这里定义的主要有5种状态，分别对应的值是从-1到3.

他们对应着线程的running, shutdown, stop, tidying, terminated这几个状态。

结构

除了前面的几个部分以外，线程池里还有如下几个成员：

  1. private final BlockingQueue<Runnable> workQueue;
  2. private final ReentrantLock mainLock = new ReentrantLock();
  3. /**
  4. * Set containing all worker threads in pool. Accessed only when
  5. * holding mainLock.
  6. */
  7. private final HashSet<Worker> workers = new HashSet<Worker>();
  8. /**
  9. * Wait condition to support awaitTermination
  10. */
  11. private final Condition termination = mainLock.newCondition();
  12. /**
  13. * Tracks largest attained pool size. Accessed only under
  14. * mainLock.
  15. */
  16. private int largestPoolSize;
  17. /**
  18. * Counter for completed tasks. Updated only on termination of
  19. * worker threads. Accessed only under mainLock.
  20. */
  21. private long completedTaskCount;
  22. private volatile ThreadFactory threadFactory;
  23. /**
  24. * Handler called when saturated or shutdown in execute.
  25. */
  26. private volatile RejectedExecutionHandler handler;
  27. /**
  28. * Timeout in nanoseconds for idle threads waiting for work.
  29. * Threads use this timeout when there are more than corePoolSize
  30. * present or if allowCoreThreadTimeOut. Otherwise they wait
  31. * forever for new work.
  32. */
  33. private volatile long keepAliveTime;
  34. /**
  35. * If false (default), core threads stay alive even when idle.
  36. * If true, core threads use keepAliveTime to time out waiting
  37. * for work.
  38. */
  39. private volatile boolean allowCoreThreadTimeOut;
  40. /**
  41. * Core pool size is the minimum number of workers to keep alive
  42. * (and not allow to time out etc) unless allowCoreThreadTimeOut
  43. * is set, in which case the minimum is zero.
  44. */
  45. private volatile int corePoolSize;
  46. /**
  47. * Maximum pool size. Note that the actual maximum is internally
  48. * bounded by CAPACITY.
  49. */
  50. private volatile int maximumPoolSize;

这些部分的内容看起来比较多，实际上他们几个都是在一些方法里经常用到的。

workQueue: 一个BlockingQueue<Runnable>队列，本身的结构可以保证访问的线程安全。相当于一个排队等待队列。当我们线程池里线程达到corePoolSize的时候，一些需要等待执行的线程就放在这个队列里等待。

workers: 一个HashSet<Worker>的集合。线程池里所有可以立即执行的线程都放在这个集合里。

mainLock: 一个访问workers所需要使用的锁。从前面的workQueue, workers这两个结构我们可以看到，如果我们要往线程池里面增加执行任务或者执行完毕一个任务，都要访问到这两个结构。所以大多数情况下为了保证线程安全，就需要使用mainLock这个锁。

corePoolSize: 处于活跃状态的最少worker数目。我们一个线程池里肯定事先创建好了若干个，等来执行任务的时候直接拿去就可以跑了。那么到底要保证最初有多少个呢？就由corePoolSize这个来指定了。

maximumPoolSize：线程池最大的长度。可以设置的一个参数。在我们当前池里面的线程数到达这个数字的时候就不能再往里面加了。需要注意的是这里是我们设定的一个池最大范围。在这里可以设定的最大数字是(2^29) -1。

其他还有几个牵涉到的成员比如说RejectedExecutionHandler等，相对都比较简单一点，代码里的注释就已经能够说清楚了。

ok，有了前面这几个基本成员的说明，我们再看看他们使用的work的结构。既然执行的都是一个Worker的集合。那么在这里Worker的定义是什么样的呢？下面是Worker的定义代码：

  1. private final class Worker
  2. extends AbstractQueuedSynchronizer
  3. implements Runnable
  4. {
  5. /**
  6. * This class will never be serialized, but we provide a
  7. * serialVersionUID to suppress a javac warning.
  8. */
  9. private static final long serialVersionUID = 6138294804551838833L;
  10. /** Thread this worker is running in.  Null if factory fails. */
  11. final Thread thread;
  12. /** Initial task to run.  Possibly null. */
  13. Runnable firstTask;
  14. /** Per-thread task counter */
  15. volatile long completedTasks;
  16. /**
  17. * Creates with given first task and thread from ThreadFactory.
  18. * @param firstTask the first task (null if none)
  19. */
  20. Worker(Runnable firstTask) {
  21. this.firstTask = firstTask;
  22. this.thread = getThreadFactory().newThread(this);
  23. }
  24. /** Delegates main run loop to outer runWorker  */
  25. public void run() {
  26. runWorker(this);
  27. }
  28. // Lock methods
  29. //
  30. // The value 0 represents the unlocked state.
  31. // The value 1 represents the locked state.
  32. protected boolean isHeldExclusively() {
  33. return getState() == 1;
  34. }
  35. protected boolean tryAcquire(int unused) {
  36. if (compareAndSetState(0, 1)) {
  37. setExclusiveOwnerThread(Thread.currentThread());
  38. return true;
  39. }
  40. return false;
  41. }
  42. protected boolean tryRelease(int unused) {
  43. setExclusiveOwnerThread(null);
  44. setState(0);
  45. return true;
  46. }
  47. public void lock()        { acquire(1); }
  48. public boolean tryLock()  { return tryAcquire(1); }
  49. public void unlock()      { release(1); }
  50. public boolean isLocked() { return isHeldExclusively(); }
  51. }

这里我们可以看到Worker本身实现了Runnable接口，所以它可以当成一个线程来执行。然后也继承了AbstractQueuedSynchronizer，也可以实现一些对本身的线程同步访问。这里最重要的几个部分在于它里面定义了一个Thread thread和Runnable firstTask。看到这里，我们可能会比较奇怪，我们只是要一个可以执行的线程，这里放一个Thread和一个Runnable的变量做什么呢？在Worker的run方法里，调用的runWorker方法到底是怎么执行的呢？我们再来看看runWorker方法：

  1. final void runWorker(Worker w) {
  2. Runnable task = w.firstTask;
  3. w.firstTask = null;
  4. boolean completedAbruptly = true;
  5. try {
  6. while (task != null || (task = getTask()) != null) {
  7. w.lock();
  8. clearInterruptsForTaskRun();
  9. try {
  10. beforeExecute(w.thread, task);
  11. Throwable thrown = null;
  12. try {
  13. task.run();
  14. } catch (RuntimeException x) {
  15. thrown = x; throw x;
  16. } catch (Error x) {
  17. thrown = x; throw x;
  18. } catch (Throwable x) {
  19. thrown = x; throw new Error(x);
  20. } finally {
  21. afterExecute(task, thrown);
  22. }
  23. } finally {
  24. task = null;
  25. w.completedTasks++;
  26. w.unlock();
  27. }
  28. }
  29. completedAbruptly = false;
  30. } finally {
  31. processWorkerExit(w, completedAbruptly);
  32. }
  33. }

这部分代码看起来挺多的，里面的beforeExecute,afterExecute的方法在默认的实现里是空的。在一些有特定要求的地方可以通过继承ThreadPoolExecutor提供自定义的实现。和前面的定义结合起来看，看来Worker这里也没干什么别的，就是绕了个圈执行了里面设定的firstTask。

如果我们仔细看其中的代码，还有一个需要注意的地方就是这里用了一个while循环来执行task，而跳出循环的条件则是要task为null。那么这个getTask是做了什么呢？一般来说我们给它一个线程，执行完任务不就完了吗？要它再去拿线程干嘛啊？我们看看它的实现：

  1. private Runnable getTask() {
  2. boolean timedOut = false; // Did the last poll() time out?
  3. retry:
  4. for (;;) {
  5. int c = ctl.get();
  6. int rs = runStateOf(c);
  7. // Check if queue empty only if necessary.
  8. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
  9. decrementWorkerCount();
  10. return null;
  11. }
  12. boolean timed;      // Are workers subject to culling?
  13. for (;;) {
  14. int wc = workerCountOf(c);
  15. timed = allowCoreThreadTimeOut || wc > corePoolSize;
  16. if (wc <= maximumPoolSize && ! (timedOut && timed))
  17. break;
  18. if (compareAndDecrementWorkerCount(c))
  19. return null;
  20. c = ctl.get();  // Re-read ctl
  21. if (runStateOf(c) != rs)
  22. continue retry;
  23. // else CAS failed due to workerCount change; retry inner loop
  24. }
  25. try {
  26. Runnable r = timed ?
  27. workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  28. workQueue.take();
  29. if (r != null)
  30. return r;
  31. timedOut = true;
  32. } catch (InterruptedException retry) {
  33. timedOut = false;
  34. }
  35. }
  36. }

从代码的注释里可以看到，这个方法要从workQueue里面获取task然后提交执行。也就是说我们线程池执行完了当前的任务后会主动到这个队列里来取后续等待的任务执行。如果当前线程因为超时、线程池要关闭等状态影响则可能会退出，而如果一切都正常的话，则会从workQueue里面调用poll或take方法取到当前任务。前面一大堆的判断和循环就是判断当前线程池长度是否超过maximumPoolSize以及当前状态是否要关闭了。如果长度超了或者状态不对则没必要继续去取任务执行了，需要尽快返回。

有了前面这部分的分析，我们知道Worker只不过包含了一个指向我们需要创建的Runnable对象，然后在Worker作为线程执行的时候再来运行这个Runnable里面的线程执行部分。

线程池执行流程

有了前面那部分的铺垫，我们再来回过头看线程池拿到一个任务后execute方法的执行。我们将这些代码拆开来看，这是第一部分：

  1. int c = ctl.get();
  2. if (workerCountOf(c) < corePoolSize) {
  3. if (addWorker(command, true))
  4. return;
  5. c = ctl.get();
  6. }

这里获取到当前正在执行的线程数目，如果这些线程的数目少于corePoolSize，则将该线程加入到线程池中。然后返回。

另外一部分的代码如下：

  1. if (isRunning(c) && workQueue.offer(command)) {
  2. int recheck = ctl.get();
  3. if (! isRunning(recheck) && remove(command))
  4. reject(command);
  5. else if (workerCountOf(recheck) == 0)
  6. addWorker(null, false);
  7. }

这里是假定如果前面不能直接加入到线程池Worker集合里，则加入到workQueue队列等待执行。里面的if else判断语句则是检查当前线程池的状态。如果线程池本身的状态是要关闭并清理了，我们则不能提交线程进去了。这里我们就要reject他们。所以前面我们看到的一些线程池拒绝线程执行的机制在这里也得到了验证。

最后面这部分的代码如下：

  1. else if (!addWorker(command, false))
  2. reject(command);

这里对应代码注释里的第3种情况，我们前面做了两种尝试，一个是将线程加入到workers集合或者workerQueue队列排队。在这两种情况都失败的情况下，我们尝试加入一个新的线程。如果这种情况下我们也失败了，则拒绝线程提交执行。

这里几个地方都用到了addWorker方法，而我们既然是execute方法，肯定要让线程执行起来。可是这里没有见到那个地方调用线程的start方法。那么很可能这个线程方法的具体调用就在addWorker方法里。

  1. private boolean addWorker(Runnable firstTask, boolean core) {
  2. retry:
  3. for (;;) {
  4. int c = ctl.get();
  5. int rs = runStateOf(c);
  6. // Check if queue empty only if necessary.
  7. if (rs >= SHUTDOWN &&
  8. ! (rs == SHUTDOWN &&
  9. firstTask == null &&
  10. ! workQueue.isEmpty()))
  11. return false;
  12. for (;;) {
  13. int wc = workerCountOf(c);
  14. if (wc >= CAPACITY ||
  15. wc >= (core ? corePoolSize : maximumPoolSize))
  16. return false;
  17. if (compareAndIncrementWorkerCount(c))
  18. break retry;
  19. c = ctl.get();  // Re-read ctl
  20. if (runStateOf(c) != rs)
  21. continue retry;
  22. // else CAS failed due to workerCount change; retry inner loop
  23. }
  24. }
  25. Worker w = new Worker(firstTask);
  26. Thread t = w.thread;
  27. final ReentrantLock mainLock = this.mainLock;
  28. mainLock.lock();
  29. try {
  30. // Recheck while holding lock.
  31. // Back out on ThreadFactory failure or if
  32. // shut down before lock acquired.
  33. int c = ctl.get();
  34. int rs = runStateOf(c);
  35. if (t == null ||
  36. (rs >= SHUTDOWN &&
  37. ! (rs == SHUTDOWN &&
  38. firstTask == null))) {
  39. decrementWorkerCount();
  40. tryTerminate();
  41. return false;
  42. }
  43. workers.add(w);
  44. int s = workers.size();
  45. if (s > largestPoolSize)
  46. largestPoolSize = s;
  47. } finally {
  48. mainLock.unlock();
  49. }
  50. t.start();
  51. // It is possible (but unlikely) for a thread to have been
  52. // added to workers, but not yet started, during transition to
  53. // STOP, which could result in a rare missed interrupt,
  54. // because Thread.interrupt is not guaranteed to have any effect
  55. // on a non-yet-started Thread (see Thread#interrupt).
  56. if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted())
  57. t.interrupt();
  58. return true;
  59. }

前面的嵌套for循环主要是用来判断当前线程池的状态是否可以允许继续加线程，同时也判断线程池的长度是否已经超标。当然，既然我们有一个线程加入了执行，当前运行的数量也要更新。如果没问题，则通过break retry;跳出这两个循环开始后面的正式执行。

在正式执行的时候我们创建一个Worker对象，并将mainLock加锁。保证后续执行部分是单线程执行的。在进入加锁的部分之后还需要再一次检查一下线程池的状态。这里我们将当前的线程加入到workers集合。然后我们通过t.start()方法正式执行线程。在这里一个线程才算是真正的执行起来了。

总结

前面我们看了一下线程池的执行机制。在默认的线程池实现里，它是通过一个workers集合来保持最核心活跃状态的线程组。当我们新加入线程执行任务时，则先利用这里的线程。如果这里的被占用满了之后则加入到workQueue这个队列里排队。这里面有一个重要的地方就是要经常检查当前线程池的状态，只有在运行状态的时候才可以往里面加线程，否则提交线程任务则会被拒绝。我们也要检查线程池的长度，防止提交的执行任务达到了我们设定的上限。为了保证线程的提交和执行安全，我们用一个lock来管理对线程集合workers和workerQueue的加锁控制。

在线程池中也有一些扩展点。比如在线程执行的过程中我们可以覆写beforeExecute, afterExecute方法来提供自己特定的功能。另外，当线程执行不符合条件要被丢弃或者拒绝的时候，我们也可以提供一些RejectExecutionHandler的具体实现。在系统的默认实现里已经提供了5种。

总的来说，对于一个线程池，它最核心的部分是对应一个线程运行集合和一个队列。如果我们能够保证好他们的状态、大小以及线程安全执行，那么基本上一个线程的雏形就差不多完成了。

上一篇：hdu2768-Cat vs. Dog：图论：二分匹配


下一篇：简单的三方登录SDK示例，Android Activity之间数据的传递