Java线程池

retry:
for (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);

    // 1.检查队列是否只在必要时为空（判断线程状态，且队列不为空）
    if (rs >= SHUTDOWN &&
        ! (rs == SHUTDOWN &&
           firstTask == null &&
           ! workQueue.isEmpty()))
        return false;
    //2.循环CAS增加线程个数
    for (;;) {
        int wc = workerCountOf(c);
        //2.1 如果线程个数超限则返回 false
        if (wc >= CAPACITY ||
            wc >= (core ? corePoolSize : maximumPoolSize))
            return false;
        //2.2 CAS方式增加线程个数，同时只有一个线程成功，成功跳出循环
        if (compareAndIncrementWorkerCount(c))
            break retry;
        //2.3 CAS失败，看线程池状态是否变化，变化则跳到外层，尝试重新获取线程池状态，否则内层重新CAS
        c = ctl.get();  // Re-read ctl
        if (runStateOf(c) != rs)
            continue retry;
    }
}
//3. 到这说明CAS成功了
boolean workerStarted = false;
boolean workerAdded = false;

Worker w = null;
 try {
     //4.创建worker
     w = new Worker(firstTask);
     final Thread t = w.thread;
     if (t != null) {
         //4.1、加独占锁 ，为了实现workers同步，因为可能多个线程调用了线程池的excute方法
         final ReentrantLock mainLock = this.mainLock;
         mainLock.lock();
         try {
             //4.2、重新检查线程池状态，以避免在获取锁前调用了shutdown接口
             int rs = runStateOf(ctl.get());

             if (rs < SHUTDOWN ||
                 (rs == SHUTDOWN && firstTask == null)) {
                 if (t.isAlive()) // precheck that t is startable
                     throw new IllegalThreadStateException();
                 //4.3添加任务
                 workers.add(w);
                 int s = workers.size();
                 if (s > largestPoolSize)
                     largestPoolSize = s;
                 workerAdded = true;
             }
         } finally {
             mainLock.unlock();
         }
         //4.4、添加成功之后启动任务
         if (workerAdded) {
             t.start();
             workerStarted = true;
         }
     }
 } finally {
     if (! workerStarted)
         addWorkerFailed(w);
 }
 return workerStarted;

Worker(Runnable firstTask) {
    setState(-1); // 在调用runWorker之前禁止中断
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);  //创建一个线程
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // 允许中断
    boolean completedAbruptly = true;
    try {
        //获取当前任务，从队列中获取任务
        while (task != null || (task = getTask()) != null) {
            w.lock();
            …………    
            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;
                // 统计当前Worker 完成了多少个任务
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //执行清理工作
        processWorkerExit(w, completedAbruptly);
    }
}

private Runnable getTask() {
        //poll()方法是否超时
        boolean timedOut = false; 
        //循环获取
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // 1.线程池未终止，且队列为空，返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            //工作线程数
            int wc = workerCountOf(c);

        
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

            //2.判断工作线程数是否超过最大线程数 && 超时判断 && 工作线程数大于0或队列为空
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                //从任务队列中获取线程
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                //获取成功
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

/**
    * 尝试停⽌所有正在执⾏的任务，停⽌处理等待的任务，
    * 并返回等待处理的任务列表。
    *
    * @return 从未开始执⾏的任务列表
    */
   public List<Runnable> shutdownNow() {
       List<Runnable> tasks; // ⽤于存储未执⾏的任务的列表
       final ReentrantLock mainLock = this.mainLock; // ThreadPoolExecutor的主锁
       mainLock.lock(); // 加锁以确保独占访问
       try {
           checkShutdownAccess(); // 检查是否有关闭的权限
           advanceRunState(STOP); // 将执⾏器的状态更新为STOP
           interruptWorkers(); // 中断所有⼯作线程
           tasks = drainQueue(); // 清空队列并将结果放⼊任务列表中
       } finally {
           mainLock.unlock(); // ⽆论try块如何退出都要释放锁
       }
       tryTerminate(); // 如果条件允许，尝试终⽌执⾏器
       return tasks; // 返回队列中未被执⾏的任务列表
   }

//记录线程池状态和线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//29
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 线程池状态
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;

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

测试自建线程工厂1-thread-1
测试自建线程工厂1-thread-2
测试自建线程工厂1-thread-3
测试自建线程工厂1-thread-2