public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>()); //SynchronousQueue是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素}

View Code

public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());//有界阻塞队列，默认大小 Integer.MAX_VALUE}

View Code

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);}public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue());//无界阻塞队列
    }public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);}

View Code

public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}

View Code

public ThreadPoolExecutor(int corePoolSize,//核心线程池大小int maximumPoolSize,//最大线程池大小long keepAliveTime,//线程池中超过corePoolSize数目的空闲线程最大存活时间；可以allowCoreThreadTimeOut(true)成为核心线程的有效时间TimeUnit unit,//keepAliveTime的时间单位BlockingQueue<Runnable> workQueue,//阻塞任务队列ThreadFactory threadFactory,//线程工厂RejectedExecutionHandler handler) {//当提交任务数超过maxmumPoolSize+workQueue之和时，任务会交给RejectedExecutionHandler来处理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;}

View Code

public class ExecutorFactory implements ThreadFactory {private String name;private int consequence=0;public Thread newThread(Runnable r) {Thread thread = new Thread(r);thread.setName(name +"-"+ consequence);consequence ++;return thread;}public ExecutorFactory() {}public ExecutorFactory(String prefix) {this.name=prefix;}
}

View Code

 public static class AbortPolicy implements RejectedExecutionHandler {public AbortPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {throw new RejectedExecutionException("Task " + r.toString() +" rejected from " +e.toString());}}

View Code

 public static class CallerRunsPolicy implements RejectedExecutionHandler {public CallerRunsPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {r.run();}}}

View Code

 public static class DiscardOldestPolicy implements RejectedExecutionHandler {public DiscardOldestPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {e.getQueue().poll();e.execute(r);}}}

View Code

  public static class DiscardPolicy implements RejectedExecutionHandler {public DiscardPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}}

View Code

public class MyRejectPolicy implements RejectedExecutionHandler{  public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {  //Sender是我的Runnable类，里面有message字段  if (r instanceof Sender) {  Sender sender = (Sender) r;  //直接打印
            System.out.println(sender.getMessage());  }  }
}

View Code

    public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newFixedThreadPool(10);Future<String> f1 = executorService.submit(new Callable<String>() {@Overridepublic String call() throws Exception {return "hello";}});System.out.println(f1.get());}

   public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;}

 public <T> Future<T> submit(Runnable task, T result) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task, result);execute(ftask);return ftask;}

 public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;}

 protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {return new FutureTask<T>(runnable, value);}

  protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable);}

//任务队列
private final BlockingQueue<Runnable> workQueue;
//作业线程集合
private final HashSet<Worker> workers = new HashSet<Worker>();

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();//判断当前线程池的线程数是否少于核心线程数，只要少于核心线程数都会addWorker创建一个新Worker(新线程)//来处理新任务if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}//当前线程数大于核心线程数或者addWroker失败，需要把任务提交到任务队列，等待Worker线程空闲后处理if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();//判断当前线程池状态是否正在运行（防止前面判断时候出现并发问题）if (! isRunning(recheck) && remove(command))reject(command);//如果当前线程池数量为0则创建新线程。else if (workerCountOf(recheck) == 0)addWorker(null, false);}//执行到这里代表当前线程已超越了核心线程且任务提交到任务队列失败。（可以注意这里的addWorker是false）//那么这里再次调用addWroker创建新线程（这时创建的线程是maximumPoolSize）。//如果还是提交任务失败则调用reject处理失败任务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;//cas修改增加线程池所拥有的线程数if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (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 {final ReentrantLock mainLock = this.mainLock;//创建一个新的Worker，则封装了一个firstTaskw = new Worker(firstTask);final Thread t = w.thread;if (t != null) {mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();int rs = runStateOf(c);if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();//提交Worker到线程池所维护的workers集合中（可以认为这是一组线程）
                        workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {//启动线程，执行Worker执行的run实现
                    t.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 runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}

 public void run() {runWorker(this);}

final void runWorker(Worker w) {/**这里获取当前执行线程，就是Worker所封装的Thread（因为是通过该Thread启动的，然后执行自身的run方法）**/Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;// 这个似乎没什么用的w.unlock(); // allow interruptsboolean completedAbruptly = true;try {//如果存在第一个任务则直接执行该任务，否则从任务队列里阻塞获取任务while (task != null || (task = getTask()) != null) {w.lock();//如果线程池状态已被标为停止，那么则不允许该线程继续执行任务！或者该线程已是中断状态,//也不允许执行任务,还需要中断该线程!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 {//执行到这里代表该线程已被终止，将被回收（从线程池的workers里删除该线程）。//这个方法同时也代表了当线程超出了空闲时间后，将不再由线程池维护，而是被GC回收。具体可以看   //getTask。由于getTask是以阻塞方式从阻塞队列获取任务，可以通过阻塞获取时候设定一个阻塞时间//来达到 keepAliveTime空闲功能。
            processWorkerExit(w, completedAbruptly);}}

private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?retry:for (;;) {int c = ctl.get();//获取当前线程池的状态（这部分具体最后讲）int rs = runStateOf(c);// 如果线程池已被shutdown或者由于其他原因关闭，那么则终止该线程，返回null，最后就会走//processWorkerExit方法 了if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}boolean timed;      // Are workers subject to culling?for (;;) {//获取线程池当前的线程数（worker数量则代表线程数）int wc = workerCountOf(c);//判断是否需要采取设置 阻塞时间的方式获取任务.如果核心线程也需要空闲回收或者当前线程数//量已经超越了核心线程数，那么都需要采取阻塞时间获取任务方式。timed = allowCoreThreadTimeOut || wc > corePoolSize;//判断是否需要跳出循环，循环仅仅只是为了cas修改减少线程池的线程数。if (wc <= maximumPoolSize && ! (timedOut && timed))break;// 执行到这里代表阻塞获取任务超时，keepAlivetime时间到了。该线程将被回收if (compareAndDecrementWorkerCount(c))return null;c = ctl.get();  // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop
            }try {//如果需要采用阻塞形式获取，那么就poll设定阻塞时间，否则take无限期等待。Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock(); //上锁try {//判断调用者是否有权限shutdown线程池
        checkShutdownAccess();//CAS+循环设置线程池状态为shutdown
        advanceRunState(SHUTDOWN);//中断所有空闲线程
        interruptIdleWorkers();onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {mainLock.unlock(); //解锁
    }//尝试终止线程池
    tryTerminate();
}

 private void interruptIdleWorkers() {interruptIdleWorkers(false);}

 private void interruptIdleWorkers(boolean onlyOne) {// onlyOne如果为true，最多interrupt一个workerfinal ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;if (!t.isInterrupted() && w.tryLock()) {try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}}