java 中ThreadPoolExecutor原理分析

/**   * @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  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;  }

public void execute(Runnable command) {    int c = ctl.get();    // 如果当期数量小于corePoolSize    if (workerCountOf(c) < corePoolSize) {      // 尝试增加worker      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);      // 如果工作线程都down了      else if (workerCountOf(recheck) == 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.      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) {          // 如果添加成功，则启动该线程，执行Worker的run方法，Worker的run方法执行外部的runWorker(Worker)          t.start();          workerStarted = true;        }      }    } finally {      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. */    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) {      setState(-1); // inhibit interrupts until runWorker      this.firstTask = firstTask;      this.thread = getThreadFactory().newThread(this);    }    /** Delegates main run loop to outer runWorker */    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;    w.unlock(); // allow interrupts    boolean completedAbruptly = true;    try {      while (task != null || (task = getTask()) != null) {        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 = null;          w.completedTasks++;          w.unlock();        }      }      completedAbruptly = false;    } finally {      processWorkerExit(w, completedAbruptly);    }  }

public V get() throws InterruptedException, ExecutionException {    int s = state;    if (s <= COMPLETING)      s = awaitDone(false, 0L);    return report(s);  }

static final class WaitNode {    volatile Thread thread;    volatile WaitNode next;    WaitNode() { thread = Thread.currentThread(); }  }private int awaitDone(boolean timed, long nanos)    throws InterruptedException {    final long deadline = timed ? System.nanoTime() + nanos : 0L;    WaitNode q = null;    boolean queued = false;    for (;;) {      if (Thread.interrupted()) {        removeWaiter(q);        throw new InterruptedException();      }      int s = state;      if (s > COMPLETING) {        if (q != null)          q.thread = null;        return s;      }      else if (s == COMPLETING) // cannot time out yet        Thread.yield();      else if (q == null)        q = new WaitNode();      else if (!queued)        queued = UNSAFE.compareAndSwapObject(this, waitersOffset,                           q.next = waiters, q);      else if (timed) {        nanos = deadline - System.nanoTime();        if (nanos <= 0L) {          removeWaiter(q);          return state;        }        LockSupport.parkNanos(this, nanos);      }      else        LockSupport.park(this);    }  }

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);        }        if (ran)          set(result);      }    } finally {      // runner must be non-null until state is settled to      // prevent concurrent calls to run()      runner = null;      // state must be re-read after nulling runner to prevent      // leaked interrupts      int s = state;      if (s >= INTERRUPTING)        handlePossibleCancellationInterrupt(s);    }  }