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