vlambda博客
学习文章列表

/**
 * The main pool control state, ctl, is an atomic integer packing
 * two conceptual fields
 * workerCount, indicating the effective number of threads
 * runState, indicating whether running, shutting down etc
 *
 * The runState provides the main lifecycle control, taking on values:看到英文就头晕？没事啦~

@ReachabilitySensitive
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;

private final class Worker
     extends AbstractQueuedSynchronizer
     implements Runnable

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

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.
 */
     int c = ctl.get();
     if (workerCountOf(c) < corePoolSize) {
         if (addWorker(command, true))
             return;
        c = ctl.get();
     }
     /**
     * 2、如果线程池RUNNING状态，且入队列成功
     */
     if (isRunning(c) && workQueue.offer(command)) {
         int recheck = ctl.get();

     //如果再次校验过程中，线程池不是RUNNING状态，
     // 并且remove(command)--workQueue.remove()成功，拒绝当前command
         if (! isRunning(recheck) && remove(command))
             reject(command);

         //为什么只检查运行的worker数量是不是0呢？？为什么不和corePoolSize比较呢？？
         // 只保证有一个worker线程可以从queue中获取任务执行就行了？？
         // 因为只要还有活动的worker线程，就可以消费workerQueue中的任务
         else if (workerCountOf(recheck) == 0)
             addWorker(null, false);
         }
         /**
         * 3、如果线程池不是running状态 或者 无法入队列
         * 尝试开启新线程，扩容至maxPoolSize，
         * 如果addWork(command, false) 失败了，拒绝当前command
         */
         else if (!addWorker(command, false))
             reject(command);
         }}