public static ExecutorService newFixedThreadPool(int nThreads) {  return new ThreadPoolExecutor(nThreads, nThreads,  0L, TimeUnit.MILLISECONDS,  new LinkedBlockingQueue<Runnable>());
}  

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {  return new ThreadPoolExecutor(nThreads, nThreads,  0L, TimeUnit.MILLISECONDS,  new LinkedBlockingQueue<Runnable>(),  threadFactory);
}  

public ThreadPoolExecutor(int corePoolSize,  int maximumPoolSize,  long keepAliveTime,  TimeUnit unit,  BlockingQueue<Runnable> workQueue) {  this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,  Executors.defaultThreadFactory(), defaultHandler);
}

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

ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue
RejectedExecutionHandler handler)corePoolSize: 线程池维护线程的最少线程数,也是核心线程数,包括空闲线程
maximumPoolSize: 线程池维护线程的最大线程数
keepAliveTime: 线程池维护线程所允许的空闲时间
unit: 程池维护线程所允许的空闲时间的单位
workQueue: 线程池所使用的缓冲队列
handler: 线程池对拒绝任务的处理策略

public Future<?> submit(Runnable task) {  if (task == null) throw new NullPointerException();  RunnableFuture<Void> ftask = newTaskFor(task, null);  execute(ftask);//执行任务  return ftask;
}  /** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException       {@inheritDoc} */
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;
}  /** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException       {@inheritDoc} */
public <T> Future<T> submit(Callable<T> task) {  if (task == null) throw new NullPointerException();  RunnableFuture<T> ftask = newTaskFor(task);  execute(ftask);//执行任务  return ftask;
}  

public FutureTask(Runnable runnable, V result) {  this.callable = Executors.callable(runnable, result);  this.state = NEW;       // ensure visibility of callable
}  

public static <T> Callable<T> callable(Runnable task, T result) {  if (task == null)  throw new NullPointerException();  return new RunnableAdapter<T>(task, result);
}  

static final class RunnableAdapter<T> implements Callable<T> {  final Runnable task;  final T result;  RunnableAdapter(Runnable task, T result) {  this.task = task;  this.result = result;  }  public T call() {  task.run();  return result;  }
}  

在submit中最重要的是execute这个方法，这个方法也是我们分析的重点

execute方法：

public void execute(Runnable command) {  if (command == null)  throw new NullPointerException();  int c = ctl.get();  if (workerCountOf(c) < corePoolSize) {//  if (addWorker(command, true))  return;  c = ctl.get();  }  if (isRunning(c) && workQueue.offer(command)) {  int recheck = ctl.get();  if (! isRunning(recheck) && remove(command))  reject(command);  else if (workerCountOf(recheck) == 0)  addWorker(null, false);  }  else if (!addWorker(command, false))  reject(command);
}  

1、如果少于corePoolSize数量的线程在运行，则启动一个新的线程并把传进来的Runnable做为第一个任务。然后会检查线程的运行状态和worker的数量，阻止不符合要求的任务添加到线程中

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();  workerStarted = true;  }  }
} finally {  if (! workerStarted)  addWorkerFailed(w);
}  

Java中的线程池

我们一般将任务(Task)提交到线程池中运行，对于一个线程池而言，需要关注的内容有以下几点：
在什么样的线程中执行任务
任务按照什么顺序来执行(FIFO,LIFO,优先级)
最多有多少个任务能并发执行
最多有多个任务等待执行
如果系统过载则需要拒绝一个任务，如何通知任务被拒绝？
在执行一个任务之前或之后需要进行哪些操作
围绕上面的问题，我们来研究一下java中的线程池

线程池的创建

Exectors.newFixedThreadPool(int size)：创建一个固定大小的线程池。 每来一个任务创建一个线程，当线程数量为size将会停止创建。当线程池中的线程已满，继续提交任务，如果有空闲线程那么空闲线程去执行任务，否则将任务添加到一个无界的等待队列中。
Exectors.newCachedThreadPool():创建一个可缓存的线程池。对线程池的规模没有限制，当线程池的当前规模超过处理需求时(比如线程池中有10个线程，而需要处理的任务只有5个)，那么将回收空闲线程。当需求增加时则会添加新的线程。
Exectors.newSingleThreadExcutor():创建一个单线程的Executor,它创建单个工作者线程来执行任务，如果这个线程异常结束，它会创建另一个线程来代替。
Exectors.newScheduledThreadPool():创建一个固定长度的线程池，而且以延迟或定时的方式来执行任务。
上面都是通过工厂方法来创建线程池，其实它们内部都是通过创建ThreadPoolExector对象来创建线程池的。下面是ThreadPoolExctor的构造函数。

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {...
}

我们看到构造函数是public类型的，所以我们也可以自定义自己的线程池。

在什么样的线程中执行任务？

java中对于任务的描述有两种，一种是Runnable型的任务，一种是Callable型的任务。前者运行结束后不会返回任何东西，而后者可以返回我们需要的计算结果，甚至异常。

在没有返回值的线程中运行

创建一个线程池，然后调用其execute方法，并将一个Runnable对象传递进去即可。

ExectorService exector = Exectors.newCachedThreadPool();
exector.execute(new Runnable(){
public void run(){
System.out.println("running...");
}
});

在有返回值的线程中运行

ExectorService exector = Exectors.newCachedThreadPool();
Callable<Result> task = new Callable<Result>() {public Result call() {return new Computor().compute();}
};
Future<Result> future = exector.submit(task);
result = future.get();  //改方法会一直阻塞，直到提交的任务被运行完毕

任务按照什么顺序来执行(FIFO,优先级)

如果任务按照某种顺序来执行的话，则任务一定是串行执行的。我们可以看到在ThreadPoolExecutor中第四个参数是BlockingQueue,提交的任务都先放到该队列中。如果传入不同的BlockQueue就可以实现不同的执行顺序。传入LinkedBlockingQueue则表示先来先服务，传入PriorityBlockingQueue则使用优先级来处理任务

Exectors.newSingleThreadExcutor()使用的是先来先服务策略

最多有多少个任务能并发执行

线程池中的线程会不断从workQueue中取任务来执行，如果没任务可执行，则线程处于空闲状态。
在ThreadPoolExecutor中有两个参数corePoolSize和maximumPoolSize，前者被称为基本大小，表示一个线程池初始化时，里面应该有的一定数量的线程。但是默认情况下，ThreadPoolExecutor在初始化是并不会马上创建corePoolSize个线程对象，它使用的是懒加载模式。

• 当线程数小于corePoolSize时,提交一个任务创建一个线程(即使这时有空闲线程)来执行该任务。
• 当线程数大于等于corePoolSize，首选将任务添加等待队列workQueue中（这里的workQueue是上面的BlockingQueue），等有空闲线程时，让空闲线程从队列中取任务。
• 当等待队列满时，如果线程数量小于maximumPoolSize则创建新的线程，否则使用拒绝线程处理器来处理提交的任务。

最多有多少的任务等待执行

这个问题和BlockingQueue相关。 BlockingQueue有三个子类，一个是ArrayBlockingQueue(有界队列),一个是LinkedBlockingQueue(默认无界，但可以配置为有界)，PriorityBlockingQueue(默认无界，可配置为有界)。所以，对于有多少个任务等待执行与传入的阻塞队列有关。

newFixedThreadPool和newSingleThreadExector使用的是LinkedBlockingQueue的无界模式。而newCachedThreadPool使用的是SynchronousQueue，这种情况下线程是不需要排队等待的，SynchronousQueue适用于线程池规模无界。

如果系统过载则需要拒绝一个任务，如何通知任务被拒绝？

当有界队列被填满或者某个任务被提交到一个已关闭的Executor时将会启动饱和策略，即使用RejectedExecutionHandler来处理。JDK中提供了几种不同的RejectedExecutionHandler的实现：AbortPolicy，CallerRunsPolicy, DiscardPolicy和DiscardOldestPolicy。

AbortPolicy：默认的饱和策略。该策略将抛出未检查的RejectedExcutionException,调用者可以捕获这个异常，然后根据自己的需求来处理。

DiscardPolicy：该策略将会抛弃提交的任务

DiscardOldestPolicy：该策略将会抛弃下一个将被执行的任务(处于队头的任务)，然后尝试重新提交该任务到等待队列

CallerRunsPolicy:该策略既不会抛弃任务也不会抛出异常，而是在调用execute()的线程中运行任务。比如我们在主线程中调用了execute(task)方法，但是这时workQueue已经满了，并且也不会创建的新的线程了。这时候将会在主线程中直接运行execute中的task。

在执行一个任务之前或之后需要进行哪些操作

ThreadPoolExecutor是可扩展的，它提供了几个可以重载的方法：beforeExecute,afterExecute和terminated,这里用到了面向的切面编程的思想。无论任务是从run中正常返回，还是抛出异常而返回，afterExectue都会被调用。如果 beforeExecute中抛出了一个 RunntimeException,那么任务将不会被执行，并且 afterExecute也不会被调用。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;public class Test {public static void main(String[] args) {TimingThreadPool executor = new TimingThreadPool(5, 10, 1,TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>());for (int i = 0; i < 5; i++)executor.execute(new Runnable() {@Overridepublic void run() {System.out.println("running1....");}});executor.shutdown();}
}class TimingThreadPool extends ThreadPoolExecutor {private final ThreadLocal<Long> startTime = new ThreadLocal<Long>();private final AtomicLong numTasks = new AtomicLong();private final AtomicLong totalTime = new AtomicLong();public TimingThreadPool(int corePoolSize, int maximumPoolSize,long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);}@Overrideprotected void beforeExecute(Thread t, Runnable r) {super.beforeExecute(t, r);startTime.set(System.nanoTime());}@Overrideprotected void afterExecute(Runnable r, Throwable t) {try {long endTime = System.nanoTime();long taskTime = endTime - startTime.get();numTasks.incrementAndGet();totalTime.addAndGet(taskTime);} finally {super.afterExecute(r, t);}}@Overrideprotected void terminated() {try {System.out.println(String.format("Terminated: arg time = %d",totalTime.get() / numTasks.get()));} finally {super.terminated();}}
}

上面的代码统计任务平均执行时间，在每个线程中beforeExecute和afertExecute都会执行一次，而terminated等线程池关闭的时候执行

参考：Java多线程和线程池

参考：java中的线程池