JAVA线程池实现原理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

1
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

1
2
3
4
5
6
7
8
9
10
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// Packing and unpacking ctl

// 计算当前运行状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 计算当前线程数量
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 通过状态和线程数生成ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

1
2
3
4
5
6
7
8
9
private static final int COUNT_BITS = Integer.SIZE - 3;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/**
* Core pool size is the minimum number of workers to keep alive
* (and not allow to time out etc) unless allowCoreThreadTimeOut
* is set, in which case the minimum is zero.
*/
// 核心池大小，若allowCoreThreadTimeOut被设置，核心线程全部空闲超时被回收的情况下会为0
private volatile int corePoolSize;

/**
* Maximum pool size. Note that the actual maximum is internally
* bounded by CAPACITY.
*/
// 最大池大小，不得超过CAPACITY
private volatile int maximumPoolSize;

//线程池的最大容量,其值的二进制为:00011111111111111111111111111111（29个1）
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 通过与的方式，获取ctl的低29位，也就是线程池中工作线程的数量
private static int workerCountOf(int c)  { return c & CAPACITY; }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public <T> Future<T> submit(Callable<T> task) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<T> ftask = newTaskFor(task);
  execute(ftask);
  return ftask;
}

public Future<?> submit(Runnable task) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<Void> ftask = newTaskFor(task, null);
  execute(ftask);
  return ftask;
}

public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public void execute(Runnable command) {
  if (command == null)
      throw new NullPointerException();
  // 获取线程池控制状态ctl的值
  int c = ctl.get();

  //worker数量小于corePoolSize
  if (workerCountOf(c) < corePoolSize) {
    // 创建worker,addWorker方法boolean参数用来判断是否创建核心线程
    // true比较的是 corePoolSize
    // false比较的是 maximumPoolSize
    if (addWorker(command, true))
      // 成功则返回
      return;
    // 失败则再次获取线程池控制状态
    c = ctl.get();
  }

  // 线程池处于RUNNING状态且线程数大于核心线程数
  // 将任务加入workQueue任务缓存队列
  if (isRunning(c) && workQueue.offer(command)) {
    // 再次检查，获取线程池控制状态
    // 防止在任务入队的过程中线程池关闭了或者线程池中没有线程了
    int recheck = ctl.get();

    //线程池不处于RUNNING状态，且将任务从workQueue移除成功
    if (! isRunning(recheck) && remove(command))
      //采取任务拒绝策略
      reject(command);

    //worker数量等于0  
    else if (workerCountOf(recheck) == 0)
        addWorker(null, false);
  }
  else if (!addWorker(command, false))
      reject(command);
}

1
private final BlockingQueue<Runnable> workQueue;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
private Runnable getTask() {
  // 标识当前线程是否超时未能获取到task对象
  boolean timedOut = false; // Did the last poll() time out?

  for (;;) {
    // 获取线程池的控制状态
    int c = ctl.get();

    // 获取线程池的运行状态
    int rs = runStateOf(c);

    // 如果线程池状态大于等于STOP，或者处于SHUTDOWN状态，并且阻塞队列为空
    // 线程池工作线程数量递减，方法返回null，回收线程
    // Check if queue empty only if necessary.
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
      decrementWorkerCount();
      return null;
    }

    // 获取worker数量
    int wc = workerCountOf(c);

    // 标识当前线程在空闲时，是否应该超时回收
    // 如果allowCoreThreadTimeOut为ture，或当前线程数大于核心池大小，则需要超时回收
    // Are workers subject to culling?
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

    // 如果worker数量大于maximumPoolSize(有可能调用了 setMaximumPoolSize()
    // 导致worker数量大于maximumPoolSize)
    if ((wc > maximumPoolSize || (timed && timedOut))
        && (wc > 1 || workQueue.isEmpty())) {
      if (compareAndDecrementWorkerCount(c))
          return null;
      continue;
    }

    try {

      // 如果允许超时回收，则调用阻塞队列的poll()
      // 只在keepAliveTime时间内等待获取任务，一旦超过则返回null
      // 否则调用take()，如果队列为空，线程进入阻塞状态
      // 无限时等待任务，直到队列中有可取任务或者响应中断信号退出
      Runnable r = timed ?
        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
        workQueue.take();

      // 若task不为null，则返回成功获取的task对象  
      if (r != null)
          return r;

      // 若返回task为null，表示线程空闲时间超时，则设置timeOut为true   
      timedOut = true;
    } catch (InterruptedException retry) {
      // 如果此worker发生了中断，采取的方案是重试，没有超时
      // 在哪些情况下会发生中断？调用setMaximumPoolSize()，shutDown()，shutDownNow()
      timedOut = false;
    }
  }
}

1
2
3
public interface RejectedExecutionHandler {
  void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable{

  /** Thread this worker is running in.  Null if factory fails. */
  // Worker持有的线程
  final Thread thread;
  /** Initial task to run.  Possibly null. */
  // 初始化的任务，可以为null
  Runnable firstTask;
  /** Per-thread task counter */
  // 线程任务计数器（当前线程完成的任务数）
  volatile long completedTasks;

  /**
  * Creates with given first task and thread from ThreadFactory.
  * @param firstTask the first task (null if none)
  */
  Worker(Runnable firstTask) {

    //设置AQS的state为-1，在执行runWorker()方法之前阻止线程中断
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;

    //利用指定的线程工厂创建一个线程，注意，参数是Worker实例本身this
    //也就是当执行start方法启动线程thread时，真正执行的是Worker类的run方法
    this.thread = getThreadFactory().newThread(this);
  }
}

1
2
3
4
5
/**
* Set containing all worker threads in pool. Accessed only when
* holding mainLock.
*/
private final HashSet<Worker> workers = new HashSet<Worker>();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
// 是否持有独占锁，重写AQS方法
@Override
protected boolean isHeldExclusively() {
  return getState() != 0;
}

// 尝试获取锁，重写AQS方法
@Override
protected boolean tryAcquire(int unused) {
    if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
    }
    return false;
}

// 尝试释放锁，重写AQS方法
@Override
protected boolean tryRelease(int unused) {
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
}

// 获取锁
public void lock()        { acquire(1); }

// 尝试获取锁
public boolean tryLock()  { return tryAcquire(1); }

// 释放锁
public void unlock()      { release(1); }

// 是否持有锁
public boolean isLocked() { return isHeldExclusively(); }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
private boolean addWorker(Runnable firstTask, boolean core) {
  retry:
  for (;;) {
      int c = ctl.get();
      int rs = runStateOf(c);

      // Check if queue empty only if necessary.
      if (rs >= SHUTDOWN &&
          ! (rs == SHUTDOWN &&
              firstTask == null &&
              ! workQueue.isEmpty()))
          return false;

      for (;;) {
          int wc = workerCountOf(c);
          if (wc >= CAPACITY ||
              wc >= (core ? corePoolSize : maximumPoolSize))
              return false;
          if (compareAndIncrementWorkerCount(c))
              break retry;
          c = ctl.get();  // Re-read ctl
          if (runStateOf(c) != rs)
              continue retry;
          // else CAS failed due to workerCount change; retry inner loop
      }
  }

  boolean workerStarted = false;
  boolean workerAdded = false;
  Worker w = null;
  try {
      w = new Worker(firstTask);
      final Thread t = w.thread;
      if (t != null) {
          final ReentrantLock mainLock = this.mainLock;
          mainLock.lock();
          try {
              // Recheck while holding lock.
              // Back out on ThreadFactory failure or if
              // shut down before lock acquired.
              int rs = runStateOf(ctl.get());

              if (rs < SHUTDOWN ||
                  (rs == SHUTDOWN && firstTask == null)) {
                  if (t.isAlive()) // precheck that t is startable
                      throw new IllegalThreadStateException();
                  workers.add(w);
                  int s = workers.size();
                  if (s > largestPoolSize)
                      largestPoolSize = s;
                  workerAdded = true;
              }
          } finally {
              mainLock.unlock();
          }
          if (workerAdded) {
              t.start();
              workerStarted = true;
          }
      }
  } finally {
      if (! workerStarted)
          addWorkerFailed(w);
  }
  return workerStarted;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
final void runWorker(Worker w) {
  // 获取当前线程
  Thread wt = Thread.currentThread();

  //获取w的firstTask
  Runnable task = w.firstTask;

  //设置w的firstTask为null
  w.firstTask = null;

  // 释放锁，设置AQS的state为0，允许中断
  w.unlock(); // allow interrupts

  //用于标识线程是否异常终止，finally中processWorkerExit()方法会有不同逻辑
  boolean completedAbruptly = true;
  try {
    // 自旋，循环调用getTask()获取任务,不断从任务缓存队列获取任务并执行
    while (task != null || (task = getTask()) != null) {
      //进入循环内部，代表已经获取到可执行的任务
      // 则对worker对象加锁，保证线程在执行任务过程中不会被中断
      w.lock();

      // If pool is stopping, ensure thread is interrupted; 
      // 若线程池状态大于等于STOP，那么意味着该线程要中断
      // if not, ensure thread is not interrupted.  This
      // requires a recheck in second case to deal with
      // shutdownNow race while clearing interrupt
      if ((runStateAtLeast(ctl.get(), STOP) ||
            (Thread.interrupted() && // 线程被中断
            runStateAtLeast(ctl.get(), STOP))) && //且是因为线程池内部状态变化而被中断
          !wt.isInterrupted())  //确保该线程未被中断
        // 发出中断请求
        wt.interrupt();
      try {
        // 开始执行任务前的Hook方法
        beforeExecute(wt, task);
        Throwable thrown = null;
        try {
          // 到这里正式开始执行任务
          task.run();
        } catch (RuntimeException x) {
          thrown = x; throw x;
        } catch (Error x) {
          thrown = x; throw x;
        } catch (Throwable x) {
          thrown = x; throw new Error(x);
        } finally {
          // 执行任务后的Hook方法
          afterExecute(task, thrown);
        }
      } finally {
        // 置空task，准备通过getTask()获取下一个任务
        task = null;
        // 计数+1
        w.completedTasks++;
        // 释放掉worker持有的独占锁
        w.unlock();
      }
    }
    completedAbruptly = false;
  } finally {
    // 获取不到任务时，主动回收自己
    // 线程执行结束可能有两种情况：
    // 1.getTask()返回null，也就是说，这个worker的使命结束了，线程执行结束
    // 2.任务执行过程中发生了异常
    // 第一种情况，getTask()返回null，那么getTask()中会将workerCount递减
    // 第二种情况，workerCount没有进行处理，这个递减操作会在processWorkerExit()中处理
    processWorkerExit(w, completedAbruptly);
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 private void processWorkerExit(Worker w, boolean completedAbruptly) {

  // 如果用户任务执行过程中发生了异常，则需要递减workerCount 
  if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
    decrementWorkerCount();

  final ReentrantLock mainLock = this.mainLock;
  
  //获取全局锁
  mainLock.lock();
  try {
    // 将worker完成任务的数量累加到总的完成任务数中
    completedTaskCount += w.completedTasks;

    // 从workers集合中移除该worker
    workers.remove(w);
  } finally {
    mainLock.unlock();
  }

  // 尝试终止线程池
  tryTerminate();

  // 获取线程池控制状态
  int c = ctl.get();

  // 线程池运行状态小于STOP
  if (runStateLessThan(c, STOP)) {

    // 如果用户任务执行过程中发生了异常，则直接调用addWorker()方法创建线程
    if (!completedAbruptly) {

      // 是否允许核心线程超时
      int min = allowCoreThreadTimeOut ? 0 : corePoolSize;

      // 允许核心超时并且workQueue阻塞队列不为空，那线程池中至少有一个工作线程
      if (min == 0 && ! workQueue.isEmpty())
        min = 1;

      // 如果工作线程数量workerCount大于等于核心池大小corePoolSize，
      // 或者允许核心超时并且workQueue阻塞队列不为空时
      // 线程池中至少有一个工作线程，直接返回  
      if (workerCountOf(c) >= min)
        return; // replacement not needed
    }
    // 不满足上述条件，则调用addWorker()方法创建线程
    // 创建新的线程取代当前线程
    addWorker(null, false);
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
final void tryTerminate() {
  for (;;) {
    int c = ctl.get();
    if (isRunning(c) || // 线程池的运行状态为RUNNING
      runStateAtLeast(c, TIDYING) || // 线程池的运行状态大于等于TIDYING
      (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) //线程池的运行状态为SHUTDOWN且阻塞队列不为空
      return;

    // 只有当线程池的运行状态为STOP
    // 或线程池运行状态为SHUTDOWN且阻塞队列为空时，可以执行到这里
    // 如果线程池工作线程的数量不为0  
    if (workerCountOf(c) != 0) { // Eligible to terminate
      // 仅仅中断一个空闲的worker
      interruptIdleWorkers(ONLY_ONE);
      return;
    }

    // 只有当线程池工作线程的数量为0时可以执行到这里
    final ReentrantLock mainLock = this.mainLock;

    // 获取全局锁
    mainLock.lock();
    try {
      // CAS操作设置线程池运行状态为TIDYING，工作线程数量为0
      if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
          try {
            // 执行terminated()钩子方法
            terminated();
          } finally {
            // 设置线程池运行状态为TERMINATED，工作线程数量为0
            ctl.set(ctlOf(TERMINATED, 0));

            // 唤醒在termination条件上等待的所有线程
            termination.signalAll();
          }
          return;
      }
    } finally {
      mainLock.unlock();
    }
    // else retry on failed CAS
    // 若CAS操作失败则重试
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    // 检查shutdown权限
    checkShutdownAccess();

    // 设置线程池运行状态为SHUTDOWN
    advanceRunState(SHUTDOWN);

    // 中断所有空闲worker
    interruptIdleWorkers();

    // 用onShutdown()钩子方法
    onShutdown(); // hook for ScheduledThreadPoolExecutor
  } finally {
    mainLock.unlock();
  }
  tryTerminate();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
private void interruptIdleWorkers() {
  interruptIdleWorkers(false);
}

// onlyOne标识是否只中断一个线程
private void interruptIdleWorkers(boolean onlyOne) {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {

    // 遍历workers工作线程集合
    for (Worker w : workers) {
      Thread t = w.thread;

      // 线程未被中断且成功获得锁
      if (!t.isInterrupted() && w.tryLock()) {
        try {
          // 发出中断请求
          t.interrupt();
        } catch (SecurityException ignore) {
        } finally {
          // 释放锁
          w.unlock();
        }
      }
      // 若只中断一个线程，则跳出循环
      if (onlyOne)
        break;
    }
  } finally {
    // 释放锁
    mainLock.unlock();
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public List<Runnable> shutdownNow() {
  List<Runnable> tasks;
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    // 检查shutdown权限
    checkShutdownAccess();

    // 设置线程池运行状态为STOP
    advanceRunState(STOP);

    // 中断所有worker
    interruptWorkers();

    // 将任务缓存队列中等待执行的任务取出并放到list中
    tasks = drainQueue();
  } finally {
    mainLock.unlock();
  }
  // 尝试终止线程池
  tryTerminate();

  // 返回任务缓存队列中等待执行的任务列表
  return tasks;
}

1
2
3
4
5
6
7
8
9
10
11
private void interruptWorkers() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    for (Worker w : workers)
      // 调用Worker类的interruptIfStarted()方法中断线程
      w.interruptIfStarted();
  } finally {
    mainLock.unlock();
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
/**
* The default rejected execution handler
* 默认的拒绝策略 AbortPolicy 丢弃任务并抛出RejectedExecutionException异常
*/
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

// 5个参数，两个默认参数 
// Executors.defaultThreadFactory()默认的线程工厂
// defaultHandler 默认的拒绝策略 AbortPolicy 
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
          Executors.defaultThreadFactory(), defaultHandler);
}

// 6个参数，一个默认参数
// defaultHandler 默认的拒绝策略 AbortPolicy 
public ThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            ThreadFactory threadFactory) {
  this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
        threadFactory, defaultHandler);
}

// 6个参数，一个默认参数
// Executors.defaultThreadFactory() 默认的线程工厂
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
          Executors.defaultThreadFactory(), handler);
}

// 7个参数，前三个构造方法最终都调用该构造方法
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
  // 参数合法性校验                                
  if (corePoolSize < 0 ||
      maximumPoolSize <= 0 ||
      maximumPoolSize < corePoolSize ||
      keepAliveTime < 0)
      throw new IllegalArgumentException();

  // 参数合法性校验 
  if (workQueue == null || threadFactory == null || handler == null)
      throw new NullPointerException();
  this.acc = System.getSecurityManager() == null ?
          null :
          AccessController.getContext();
  //初始化对应的属性
  this.corePoolSize = corePoolSize;
  this.maximumPoolSize = maximumPoolSize;
  this.workQueue = workQueue;
  this.keepAliveTime = unit.toNanos(keepAliveTime);
  this.threadFactory = threadFactory;
  this.handler = handler;
}