线程池ThreadPoolExecutor源码解读研究（JDK1.8）

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

 1 public void execute(Runnable command) {
 2         if (command == null)
 3             throw new NullPointerException();
 4         /*
 5          * Proceed in 3 steps:
 6          *
 7          * 1. If fewer than corePoolSize threads are running, try to
 8          * start a new thread with the given command as its first
 9          * task.  The call to addWorker atomically checks runState and
10          * workerCount, and so prevents false alarms that would add
11          * threads when it shouldn‘t, by returning false.
12          *
13          * 2. If a task can be successfully queued, then we still need
14          * to double-check whether we should have added a thread
15          * (because existing ones died since last checking) or that
16          * the pool shut down since entry into this method. So we
17          * recheck state and if necessary roll back the enqueuing if
18          * stopped, or start a new thread if there are none.
19          *
20          * 3. If we cannot queue task, then we try to add a new
21          * thread.  If it fails, we know we are shut down or saturated
22          * and so reject the task.
23          */
24         int c = ctl.get();
25         if (workerCountOf(c) < corePoolSize) {
26             if (addWorker(command, true))
27                 return;
28             c = ctl.get();
29         }
30         if (isRunning(c) && workQueue.offer(command)) {
31             int recheck = ctl.get();
32             if (! isRunning(recheck) && remove(command))
33                 reject(command);
34             else if (workerCountOf(recheck) == 0)
35                 addWorker(null, false);
36         }
37         else if (!addWorker(command, false))
38             reject(command);
39 }

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))//增加worker的数量
                    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是否添加成功
        Worker w = null;
        try {
            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());计算线程池状态

                    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;//worker添加成功
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();//启动刚刚添加的任务
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);//失败后执行的操作
        }
        return workerStarted;
}

public class TestThreadPool implements Runnable {
    @Override
    public void run() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public static void main(String[] args) {
        //指定3个长度的工作队列
        LinkedBlockingDeque<Runnable> workQueue=new LinkedBlockingDeque<>(3);
        //指定线程池参数：核心线程数，线程池最大线程数量，活跃时间，工作队列
        ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(4, 7, 90, 
                TimeUnit.SECONDS, workQueue);
        for (int i = 0; i < 15; i++) {
            threadPoolExecutor.execute(new Thread(new TestThreadPool(), 
                    "线程:".concat(i+"")));
            System.out.println("线程池中活跃线程数"+threadPoolExecutor.getActiveCount());
            if(workQueue.size()>0){
                System.out.println("被阻塞的线程数为："+workQueue.size());
            }
        }
    }

线程池中活跃线程数1
线程池中活跃线程数2
线程池中活跃线程数3
线程池中活跃线程数4
线程池中活跃线程数4
被阻塞的线程数为：1
线程池中活跃线程数4
被阻塞的线程数为：2
线程池中活跃线程数4
被阻塞的线程数为：3
线程池中活跃线程数5
被阻塞的线程数为：3
线程池中活跃线程数6
被阻塞的线程数为：3
线程池中活跃线程数7
被阻塞的线程数为：3
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task Thread[线程:10,5,main] rejected from java.util.concurrent.ThreadPoolExecutor@42a57993[Running, pool size = 7, active threads = 7, queued tasks = 3, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    at main.Main.main(Main.java:19)