8000字详解Thread Pool Executor

public class WorkerThread implements Runnable {
 private String command;
 public WorkerThread(String s){
 this.command=s;
 }
 @Override
 public void run() {
 System.out.println(Thread.currentThread().getName()+" Start. Command = "+command);
 processCommand();
 System.out.println(Thread.currentThread().getName()+" End.");
 }
 private void processCommand() {
 try {
 Thread.sleep(5000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 @Override
 public String toString(){
 return this.command;
 }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SimpleThreadPool {
 public static void main(String[] args) {
 ExecutorService executor = Executors.newFixedThreadPool(5);
 for (int i = 0; i < 10; i++) {
 Runnable worker = new WorkerThread("" + i);
 executor.execute(worker);
 }
 executor.shutdown(); // This will make the executor accept no new threads and finish all existing threads in the queue
 while (!executor.isTerminated()) { // Wait until all threads are finish,and also you can use "executor.awaitTermination();" to wait
 }
 System.out.println("Finished all threads");
 }
}

pool-1-thread-2 Start. Command = 1
pool-1-thread-4 Start. Command = 3
pool-1-thread-1 Start. Command = 0
pool-1-thread-3 Start. Command = 2
pool-1-thread-5 Start. Command = 4
pool-1-thread-4 End.
pool-1-thread-5 End.
pool-1-thread-1 End.
pool-1-thread-3 End.
pool-1-thread-3 Start. Command = 8
pool-1-thread-2 End.
pool-1-thread-2 Start. Command = 9
pool-1-thread-1 Start. Command = 7
pool-1-thread-5 Start. Command = 6
pool-1-thread-4 Start. Command = 5
pool-1-thread-2 End.
pool-1-thread-4 End.
pool-1-thread-3 End.
pool-1-thread-5 End.
pool-1-thread-1 End.
Finished all threads

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
public class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {
 @Override
 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
 System.out.println(r.toString() + " is rejected");
 }
}

import java.util.concurrent.ThreadPoolExecutor;
public class MyMonitorThread implements Runnable
{
 private ThreadPoolExecutor executor;
 private int seconds;
 private boolean run=true;
 public MyMonitorThread(ThreadPoolExecutor executor, int delay)
 {
 this.executor = executor;
 this.seconds=delay;
 }
 public void shutdown(){
 this.run=false;
 }
 @Override
 public void run()
 {
 while(run){
 System.out.println(
 String.format("[monitor] [%d/%d] Active: %d, Completed: %d, Task: %d, isShutdown: %s, isTerminated: %s",
 this.executor.getPoolSize(),
 this.executor.getCorePoolSize(),
 this.executor.getActiveCount(),
 this.executor.getCompletedTaskCount(),
 this.executor.getTaskCount(),
 this.executor.isShutdown(),
 this.executor.isTerminated()));
 try {
 Thread.sleep(seconds*1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 }
}

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class WorkerPool {
 public static void main(String args[]) throws InterruptedException{
 //RejectedExecutionHandler implementation
 RejectedExecutionHandlerImpl rejectionHandler = new RejectedExecutionHandlerImpl();
 //Get the ThreadFactory implementation to use
 ThreadFactory threadFactory = Executors.defaultThreadFactory();
 //creating the ThreadPoolExecutor
 ThreadPoolExecutor executorPool = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), threadFactory, rejectionHandler);
 //start the monitoring thread
 MyMonitorThread monitor = new MyMonitorThread(executorPool, 3);
 Thread monitorThread = new Thread(monitor);
        monitorThread.start();
 //submit work to the thread pool
 for(int i=0; i<10; i++){
            executorPool.execute(new WorkerThread("cmd"+i));
 }
 Thread.sleep(30000);
 //shut down the pool
        executorPool.shutdown();
 //shut down the monitor thread
 Thread.sleep(5000);
 monitor.shutdown();
 }
}

pool-1-thread-1 Start. Command = cmd0
pool-1-thread-4 Start. Command = cmd5
cmd6 is rejected
pool-1-thread-3 Start. Command = cmd4
pool-1-thread-2 Start. Command = cmd1
cmd7 is rejected
cmd8 is rejected
cmd9 is rejected
[monitor] [0/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, isTerminated: false
[monitor] [4/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, isTerminated: false
pool-1-thread-4 End.
pool-1-thread-1 End.
pool-1-thread-2 End.
pool-1-thread-3 End.
pool-1-thread-1 Start. Command = cmd3
pool-1-thread-4 Start. Command = cmd2
[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false
[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false
pool-1-thread-1 End.
pool-1-thread-4 End.
[monitor] [4/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true
[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true

public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue,
 RejectedExecutionHandler handler)

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

//这个属性是用来存放 当前运行的worker数量以及线程池状态的
//int是32位的，这里把int的高3位拿来充当线程池状态的标志位,后29位拿来充当当前运行worker的数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//存放任务的阻塞队列
private final BlockingQueue<Runnable> workQueue;
//worker的集合,用set来存放
private final HashSet<Worker> workers = new HashSet<Worker>();
//历史达到的worker数最大值
private int largestPoolSize;
//当队列满了并且worker的数量达到maxSize的时候,执行具体的拒绝策略
private volatile RejectedExecutionHandler handler;
//超出coreSize的worker的生存时间
private volatile long keepAliveTime;
//常驻worker的数量
private volatile int corePoolSize;
//最大worker的数量,一般当workQueue满了才会用到这个参数
private volatile int maximumPoolSize;

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;
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

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) { 
 //workerCountOf获取线程池的当前线程数；小于corePoolSize，执行addWorker创建新线程执行command任务
 if (addWorker(command, true))
 return;
        c = ctl.get();
 }
 // double check: c, recheck
 // 线程池处于RUNNING状态，把提交的任务成功放入阻塞队列中
 if (isRunning(c) && workQueue.offer(command)) {
 int recheck = ctl.get();
 // recheck and if necessary 回滚到入队操作前，即倘若线程池shutdown状态，就remove(command)
 //如果线程池没有RUNNING，成功从阻塞队列中删除任务，执行reject方法处理任务
 if (! isRunning(recheck) && remove(command))
 reject(command);
 //线程池处于running状态，但是没有线程，则创建线程
 else if (workerCountOf(recheck) == 0)
 addWorker(null, false);
 }
 // 往线程池中创建新的线程失败，则reject任务
 else if (!addWorker(command, false))
 reject(command);
}

private final ReentrantLock mainLock = new ReentrantLock();

private boolean addWorker(Runnable firstTask, boolean core) {
 // CAS更新线程池数量
    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) {
                t.start(); // 线程启动，执行任务(Worker.thread(firstTask).start());
                workerStarted = true;
 }
 }
 } finally {
 if (! workerStarted)
 addWorkerFailed(w);
 }
 return workerStarted;
}

 private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
 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);
 }
 // ...
 }

//运行的线程,前面addWorker方法中就是直接通过启动这个线程来启动这个worker
final Thread thread;
//当一个worker刚创建的时候,就先尝试执行这个任务
Runnable firstTask;
//记录完成任务的数量
volatile long completedTasks;
Worker(Runnable firstTask) {
 setState(-1); // inhibit interrupts until runWorker
 this.firstTask = firstTask;
 //创建一个Thread,将自己设置给他,后面这个thread启动的时候,也就是执行worker的run方法
 this.thread = getThreadFactory().newThread(this);
} 

final void runWorker(Worker w) {
 Thread wt = Thread.currentThread();
 Runnable task = w.firstTask;
 w.firstTask = null;
 w.unlock(); // allow interrupts
 boolean completedAbruptly = true;
 try {
 // 先执行firstTask，再从workerQueue中取task(getTask())
 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);
 }
}

private Runnable getTask() {
 boolean timedOut = false; // Did the last poll() time out?
 for (;;) {
 int c = ctl.get();
 int rs = runStateOf(c);
 // Check if queue empty only if necessary.
 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
 decrementWorkerCount();
 return null;
 }
 int wc = workerCountOf(c);
 // Are workers subject to culling?
 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 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;
 }
 }
}

boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();

public class Test{
 public static void main(String[] args) {
 ExecutorService es = Executors.newCachedThreadPool();
 Future<String> future = es.submit(new Callable<String>() {
 @Override
 public String call() throws Exception {
 try {
 TimeUnit.SECONDS.sleep(2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "future result";
 }
 });
 try {
 String result = future.get();
 System.out.println(result);
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
}

// submit()在ExecutorService中的定义
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
// submit方法在AbstractExecutorService中的实现
public Future<?> submit(Runnable task) {
 if (task == null) throw new NullPointerException();
 // 通过submit方法提交的Callable任务会被封装成了一个FutureTask对象。
 RunnableFuture<Void> ftask = newTaskFor(task, null);
 execute(ftask);
 return ftask;
}

/* The run state of this task, initially NEW.
    * ...
    * Possible state transitions:
    * NEW -> COMPLETING -> NORMAL
    * NEW -> COMPLETING -> EXCEPTIONAL
    * NEW -> CANCELLED
    * NEW -> INTERRUPTING -> INTERRUPTED
    */
private volatile int state;
private static final int NEW          = 0;
private static final int COMPLETING   = 1;
private static final int NORMAL       = 2;
private static final int EXCEPTIONAL  = 3;
private static final int CANCELLED    = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED  = 6;

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

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

public void shutdown() {
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 //检查是否可以关闭线程
 checkShutdownAccess();
 //设置线程池状态
 advanceRunState(SHUTDOWN);
 //尝试中断worker
 interruptIdleWorkers();
 //预留方法,留给子类实现
 onShutdown(); // hook for ScheduledThreadPoolExecutor
 } finally {
        mainLock.unlock();
 }
 tryTerminate();
}
private void interruptIdleWorkers() {
 interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 //遍历所有的worker
 for (Worker w : workers) {
 Thread t = w.thread;
 //先尝试调用w.tryLock(),如果获取到锁,就说明worker是空闲的,就可以直接中断它
 //注意的是,worker自己本身实现了AQS同步框架,然后实现的类似锁的功能
 //它实现的锁是不可重入的,所以如果worker在执行任务的时候,会先进行加锁,这里tryLock()就会返回false
 if (!t.isInterrupted() && w.tryLock()) {
 try {
 t.interrupt();
 } catch (SecurityException ignore) {
 } finally {
 w.unlock();
 }
 }
 if (onlyOne)
 break;
 }
 } finally {
        mainLock.unlock();
 }
}

public List<Runnable> shutdownNow() {
 List<Runnable> tasks;
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 checkShutdownAccess();
 //检测权限
 advanceRunState(STOP);
 //中断所有的worker
 interruptWorkers();
 //清空任务队列
        tasks = drainQueue();
 } finally {
        mainLock.unlock();
 }
 tryTerminate();
 return tasks;
}
private void interruptWorkers() {
 final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
 try {
 //遍历所有worker，然后调用中断方法
 for (Worker w : workers)
 w.interruptIfStarted();
 } finally {
        mainLock.unlock();
 }
}

ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,
 new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());

ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build();
//Common Thread Pool
ExecutorService pool = new ThreadPoolExecutor(5, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
// excute
pool.execute(()-> System.out.println(Thread.currentThread().getName()));
 //gracefully shutdown
pool.shutdown();

 <bean id="userThreadPool" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
 <property name="corePoolSize" value="10" />
 <property name="maxPoolSize" value="100" />
 <property name="queueCapacity" value="2000" />
 <property name="threadFactory" value= threadFactory />
 <property name="rejectedExecutionHandler">
 <ref local="rejectedExecutionHandler" />
 </property>
 </bean>
    //in code
    userThreadPool.execute(thread);