Java并发-线程池

/**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters.
 *
 * @param corePoolSize the number of threads to keep in the pool, even
 *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
 * @param maximumPoolSize the maximum number of threads to allow in the
 *        pool
 * @param keepAliveTime when the number of threads is greater than
 *        the core, this is the maximum time that excess idle threads
 *        will wait for new tasks before terminating.
 * @param unit the time unit for the {@code keepAliveTime} argument
 * @param workQueue the queue to use for holding tasks before they are
 *        executed.  This queue will hold only the {@code Runnable}
 *        tasks submitted by the {@code execute} method.
 * @param threadFactory the factory to use when the executor
 *        creates a new thread
 * @param handler the handler to use when execution is blocked
 *        because the thread bounds and queue capacities are reached
 * @throws IllegalArgumentException if one of the following holds:<br>
 *         {@code corePoolSize < 0}<br>
 *         {@code keepAliveTime < 0}<br>
 *         {@code maximumPoolSize <= 0}<br>
 *         {@code maximumPoolSize < corePoolSize}
 * @throws NullPointerException if {@code workQueue}
 *         or {@code threadFactory} or {@code handler} is null
 */
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.corePoolSize = corePoolSize;
  this.maximumPoolSize = maximumPoolSize;
  this.workQueue = workQueue;
  this.keepAliveTime = unit.toNanos(keepAliveTime);
  this.threadFactory = threadFactory;
  this.handler = handler;
}

class TPE extends ThreadPoolExecutor {
	// 记录Runnable任务起始执行时间
  private ConcurrentHashMap<Integer, Long> beginTimeMaps = new ConcurrentHashMap<>();
  
  @Override
  protected void beforeExecute(Thread t, Runnable r) {
    // 设置起始时间
    beginTimeMaps.put(r.hashCode(), new Date().getTime());
  }
  
  @Override
  protected void afterExecute(Runnable r, Throwable t) {
    // 获取起始时间  
    Long begin = beginTimeMaps.remove(r.hashCode());
    Long end = new Date().getTime();
    System.out.println(end-begin);
  }
  
  @Override
  protected void terminated() {
    // code
  }

  @Override
  public void shutdown() {
    // code
    super.shutdown();
  }

  @Override
  public List<Runnable> shutdownNow() {
    // code
    return super.shutdownNow();
  }
}

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 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;

// Packing and unpacking ctl
// 运行状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 工作线程数
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 运行状态和工作线程数
private static int ctlOf(int rs, int wc) { return rs | wc; }

public void execute(Runnable command) {
  if (command == null)
    throw new NullPointerException();
  /*
   * Proceed in 3 steps:
   *
   * 1. If fewer than corePoolSize threads are running, try to
   * start a new thread with the given command as its first
   * task.  The call to addWorker atomically checks runState and
   * workerCount, and so prevents false alarms that would add
   * threads when it shouldn't, by returning false.
   *
   * 2. If a task can be successfully queued, then we still need
   * to double-check whether we should have added a thread
   * (because existing ones died since last checking) or that
   * the pool shut down since entry into this method. So we
   * recheck state and if necessary roll back the enqueuing if
   * stopped, or start a new thread if there are none.
   *
   * 3. If we cannot queue task, then we try to add a new
   * thread.  If it fails, we know we are shut down or saturated
   * and so reject the task.
   */
  // 获取线程池的runState和workerCount
  int c = ctl.get();
  // 若当前线程数小于核心线程数；则新建一个Worker并执行当前任务
  if (workerCountOf(c) < corePoolSize) {
    // 新增线程(command, core) core表示新增的是否为核心线程
    if (addWorker(command, true))
      return;
    // 添加核心线程失败，重新获取线程池状态
    c = ctl.get();
  }
  // 若当前线程数不小于核心线程数并且处于运行状态，则将任务放到阻塞队列中
  if (isRunning(c) && workQueue.offer(command)) {
    // 添加队列成功后，再次获取线程池状态
    int recheck = ctl.get();
    // 线程池不处于运行状态，并且成功移除任务。则执行拒绝策略
    if (! isRunning(recheck) && remove(command))
      // 执行拒绝策略
      reject(command);
    else if (workerCountOf(recheck) == 0)
      // 如果工作线程为0(线程池无线程)，则新建一个无任务的线程
      addWorker(null, false);
  }
  // 非运行状态或放队列失败时，直接拒绝策略
  else if (!addWorker(command, false))
    // 执行拒绝策略
    reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {
  retry:
  for (;;) {
    // 线程池状态及线程池运行状态
    int c = ctl.get();
    int rs = runStateOf(c);

    // Check if queue empty only if necessary.
    // 当rs>=SHUTDOWN时 即SHUTDOWN(0)、STOP(1)、TIDYING(2)、TERMINATED(3)，此状态不再接受新任务
    // 当rs=SHUTDOWN时，此时可创建线程条件如下：
    // 1. rs==SHUTDOWN，此时不再接受新任务，但可
    // 2. firstTask为空，不可再继续提交任务
    // 3. !workQueue.isEmpty()，为空的话则不再需要新线程
    if (rs >= SHUTDOWN &&
        ! (rs == SHUTDOWN &&
           firstTask == null &&
           ! workQueue.isEmpty()))
      return false;

    for (;;) {
      // 当前工作线程数
      int wc = workerCountOf(c);
      // 当线程数达到CAPACITY，
      // 或者core为true时达到核心线程数，
      // 或者core为false时达到最大线程数，
      // 不再创建新线程
      if (wc >= CAPACITY ||
          wc >= (core ? corePoolSize : maximumPoolSize))
        return false;
      // 使用CAS增加workerCount
      if (compareAndIncrementWorkerCount(c))
        // 成功修改workerCount，跳出最外层for循环
        break retry;
      
      // 重新获取线程池状态
      c = ctl.get();  // Re-read ctl
      // 线程池状态已被修改，继续外传for循环
      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 {
    // 根据firstTask创建新worker
    w = new Worker(firstTask);
    // 拿到当前worker的线程
    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());

        // rs<SHUTDOWN即为RUNNING，线程池处于运行状态
        // rs==SHUTDOWN时，因为可以继续执行队列中的任务，故允许添加无任务的worker
        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();
          // 更新getLargestPoolSize，线程池中出现过最大线程数
          if (s > largestPoolSize)
            largestPoolSize = s;
          workerAdded = true;
        }
      } finally {
        mainLock.unlock();
      }
      if (workerAdded) {
        // worker添加成功，启动线程
        t.start();
        workerStarted = true;
      }
    }
  } finally {
    // worker启动失败，则roll back cleanly.
    if (! workerStarted)
      addWorkerFailed(w);
  }
  return workerStarted;
}

private final class Worker
  extends AbstractQueuedSynchronizer
  implements Runnable
{
  /**
   * This class will never be serialized, but we provide a
   * serialVersionUID to suppress a javac warning.
   */
  private static final long serialVersionUID = 6138294804551838833L;

  /** Thread this worker is running in.  Null if factory fails. */
  // ThreadFactory.newThread创建的线程实例
  final Thread thread;
  /** Initial task to run.  Possibly 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) {
    // 禁止中断，直到runWorker方法执行
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    // 通过ThreadFactory创建线程实例，Worker自身作为Runnable来创建线程实例
    this.thread = getThreadFactory().newThread(this);
  }

  /** Delegates main run loop to outer runWorker  */
  // t.start() 将会调用该方法
  public void run() {
    // 执行任务核心方法
    runWorker(this);
  }

  // Lock methods
  //
  // The value 0 represents the unlocked state.
  // The value 1 represents the locked state.

  protected boolean isHeldExclusively() {
    return getState() != 0;
  }

  protected boolean tryAcquire(int unused) {
    if (compareAndSetState(0, 1)) {
      setExclusiveOwnerThread(Thread.currentThread());
      return true;
    }
    return false;
  }

  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(); }

  void interruptIfStarted() {
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
      try {
        t.interrupt();
      } catch (SecurityException ignore) {
      }
    }
  }
}

final void runWorker(Worker w) {
  Thread wt = Thread.currentThread();
  Runnable task = w.firstTask;
  w.firstTask = null;
  // unlock 将state由-1置为0。
  w.unlock(); // allow interrupts
  boolean completedAbruptly = true;
  try {
    // 初始任务不为null，或者从队列中获取到了任务
    while (task != null || (task = getTask()) != null) {
      // worker加锁
      w.lock();
      // If pool is stopping, ensure thread is interrupted;
      // 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 {
        // 钩子函数，任务执行前执行
        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 {
          // 钩子函数，任务执行后执行
          afterExecute(task, thrown);
        }
      } finally {
        // 任务执行完，清空task任务
        task = null;
        // 已完成任务+1
        w.completedTasks++;
        // worker解锁
        w.unlock();
      }
    }
    // 线程正常退出
    completedAbruptly = false;
  } finally {
    // 线程退出 completedAbruptly标识正常退出/非正常退出
    processWorkerExit(w, completedAbruptly);
  }
}

private Runnable getTask() {
  boolean timedOut = false; // Did the last poll() time out?

  for (;;) {
    // 获取当前线程池状态、运行状态
    int c = ctl.get();
    int rs = runStateOf(c);

    // Check if queue empty only if necessary.
    // rs>=STOP，线程池调用`shutdownNow`后，不再处理新任务
    // rs>=SHUTDOWN，workQueue消费完后，不再接受新任务
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
      decrementWorkerCount();
      return null;
    }

    // 当前工作线程数
    int wc = workerCountOf(c);

    // Are workers subject to culling?
    // 是否存在超时校验标识
    // `allowCoreThreadTimeOut`允许核心线程超时 或 工作线程数大于核心线程数
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

    // wc>maximumPoolSize，可能调用了`setMaximumPoolSize`，修改了最大线程数
    // timed&&timedOut，当前线程需要进行超时控制，并且上次发生了超时
    // 线程数大于1，或者阻塞队列为空
    if ((wc > maximumPoolSize || (timed && timedOut))
        && (wc > 1 || workQueue.isEmpty())) {
      // 尝试扣减工作线程
      if (compareAndDecrementWorkerCount(c))
        return null;
      continue;
    }

    try {
      // 是否需要进行超时判断 timed
      // 若为true，则需要进行超时判断，通过阻塞队列的poll方法来进行超时控制，超时则返回null
      // 若为false，则通过take获取，阻塞队列直到workQueue不为空
      Runnable r = timed ?
        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
      workQueue.take();
      if (r != null)
        return r;
      // r为null，已经超时，设置标识位
      timedOut = true;
    } catch (InterruptedException retry) {
      // 发生了中断，未发生超时，设置标识位
      timedOut = false;
    }
  }
}

private void processWorkerExit(Worker w, boolean completedAbruptly) {
  // completedAbruptly是否出现异常，将workerCount减1
  if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
    decrementWorkerCount();

  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    // 已完成任务数
    completedTaskCount += w.completedTasks;
    workers.remove(w);
  } finally {
    mainLock.unlock();
  }

  tryTerminate();

  int c = ctl.get();
  // 若线程处于RUNNING或SHUTDOWN状态时，若worker异常结束，则直接添加空任务worker
  if (runStateLessThan(c, STOP)) {
    if (!completedAbruptly) {
      // 允许核心线程超时时，如果阻塞队列不为空，则至少保留一个worker
      int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
      if (min == 0 && ! workQueue.isEmpty())
        min = 1;
      // 若不允许核心线程池超时，则workerCount不少于corePoolSize
      if (workerCountOf(c) >= min)
        return; // replacement not needed
    }
    addWorker(null, false);
  }
}

public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    // 安全策略校验
    checkShutdownAccess();
    // 切换状态为SHUTDOWN
    advanceRunState(SHUTDOWN);
    // 中断空闲线程
    interruptIdleWorkers();
    // 钩子
    onShutdown(); // hook for ScheduledThreadPoolExecutor
  } finally {
    mainLock.unlock();
  }
  // 尝试结束线程池
  tryTerminate();
}

public List<Runnable> shutdownNow() {
  List<Runnable> tasks;
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    checkShutdownAccess();
    advanceRunState(STOP);
    // 中断所有线程(空闲、非空闲)
    interruptWorkers();
    // 取出队列中所有未被执行的任务
    tasks = drainQueue();
  } finally {
    mainLock.unlock();
  }
  tryTerminate();
  return tasks;
}

public static ExecutorService newFixedThreadPool(int nThreads) {
  return new ThreadPoolExecutor(nThreads, nThreads,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newSingleThreadExecutor() {
  return new FinalizableDelegatedExecutorService
    (new ThreadPoolExecutor(1, 1,
                            0L, TimeUnit.MILLISECONDS,
                            new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                60L, TimeUnit.SECONDS,
                                new SynchronousQueue<Runnable>());
}

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
  return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
  super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
        new DelayedWorkQueue());
}