源码解析Executors.newFixedThreadPool(int)

摘要：创建一个线程池，具有固定线程数，运行在共享的无界队列中。固定线程数源码如下是的实现类。线程池中允许最大的线程数。如果线程数超过了核心线程数，过量的线程在关闭前等待新任务的最大时间。处理因为线程边界和队列容量导致的堵塞。

1.Executors.newFixedThreadPool(int nThreads)：创建一个线程池，具有固定线程数，运行在共享的无界队列中。在大多数时候，线程会主动执行任务，当所有的线程都在执行任务时，有新的任务加入进来，就会进入等待队列(可以有源源不断的任务加入进来，因为是无界队列)，当有空闲的线程，等待队列中的任务就会被执行。如果有线程在执行过程中因为执行失败要关闭，新创建的线程会替失败的线程执行接下来的任务。如果想要关闭这个线程池，可以调用ExecutorService的shutDown方法。

nThreads 固定线程数

源码如下：

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

2.ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue workQueue):是ExecutorService的实现类。

corePoolSize 核心线程数，即使是空闲的时候，线程池中也会保留线程的数量。如果设置allowCoreThreadTimeOut为true，会使用keepAliveTime作为等待工作时间，默认为false。

maximumPoolSize 线程池中允许最大的线程数。

keepAliveTime 如果线程数超过了核心线程数，过量的线程在关闭前等待新任务的最大时间。

unit keepAliveTime 参数的时间单元。

workQueue 用来存放等待执行的任务，这些任务是通过execute方法提交的Runnable任务。

源码如下：

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

3.ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)

threadFactory 用来创建新线程的工厂。

handler 处理因为线程边界和队列容量导致的堵塞。

源码如下：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue 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.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

4.LinkedBlockingQueue:Executors.newFixedThreadPool()使用基于链表结点的无界队列LinkedBlockingQueue存储待执行的任务。继承了AbstractQueue类，实现了BlockingQueue接口，采用先进先出的排列方式，头结点是入队时间最长的元素，尾结点是入队时间最短的元素。新结点添加到队尾，从队头弹出结点。链表队列的特点是：跟基于数组的队列相比有更大的吞吐量，但在大多并发应用中性能会比较差。LinkedBlockingQueue可以在创建的时候传递一个容量参数，限制队列的长度，不设定的情况下，默认是Integer.MAX_VALUE。在没有超过队列边界的情况下，每次添加会自动创建链表结点。

源码如下：

 public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node(null);
}

5.DefaultThreadFactory:返回默认的线程工厂创建新线程。执行器Executor在同一个线程组中创建所有新的线程。如果存在SecurityManager，就使用SecurityManager的线程组，否则使用当前线程的线程组。每一个新创建的线程作为非守护线程，其优先级设置为Thread.NORM_PRIORITY和线程组允许的最大优先级的较小者。可以通过Thread.getName()获取线程的名称，形式如：pool-N-thread-M，N表示线程工厂的序列号，M表示线程工厂创建的线程的序列号。

源码如下：

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    ......
}

6.AbortPolicy：默认的拒绝执行处理器。抛出一个RejectedExecutionException。

源码如下：

private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();

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

7.例子

@Test
public void testFixedDemo01() {
    ExecutorService executorService = Executors.newFixedThreadPool(4);
    for(int i = 0;i < 10;i++) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + " "  +" 走啦");
            }
        });
    }
}

运行结果：

核心线程数和最大线程数都是4，创建了一个线程工厂。