Fork me on GitHub

线程池

发表于 | 分类于 ， ， | | 阅读次数: |
字数统计: 3.9k | 阅读时长 ≈ 14

　　JAVA语言为我们提供了两种基础线程池的选择：ScheduledThreadPoolExecutor和ThreadPoolExecutor。它们都实现了ExecutorService接口（注意，ExecutorService接口本身和“线程池”并没有直接关系，它的定义更接近“执行器”，而“使用线程管理的方式进行实现”只是其中的一种实现方式）。这篇文章中，我们主要围绕 ThreadPoolExecutor 类进行讲解。

ThreadPoolExecutor的构造方法
逻辑结构
1
2
3
4
5
6
7
8
9
10
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;
}

　　构造函数中需要传入的参数包括corePoolSize、maximumPoolSize、keepAliveTime、timeUnit和workQueue。要明确理解这些参数（和后续将要介绍的参数）的含义，就首先要搞清楚 ThreadPoolExecutor 线程池的逻辑结构。

　　一定要注意一个概念，即存在于线程池中容器的一定是 Thread 对象，而不是您要求运行的任务（所以叫线程池而不叫任务池也不叫对象池，更不叫游泳池）；您要求运行的任务将被线程池分配给某一个空闲的 Thread 运行。从上图中，我们可以看到构成线程池的几个重要元素：

  • 等待队列：顾名思义，就是您调用线程池对象的 submit() 方法或者 execute() 方法，要求线程池运行的任务（这些任务必须实现 Runnable 接口或者 Callable 接口）。但是出于某些原因线程池并没有马上运行这些任务，而是送入一个队列等待执行。
  • 核心线程：线程池主要用于执行任务的是“核心线程”，“核心线程”的数量是您创建线程时所设置的 corePoolSize 参数决定的。如果不进行特别的设定，线程池中始终会保持 corePoolSize 数量的线程数（不包括创建阶段）。
  • 非核心线程：一旦任务数量过多（由等待队列的特性决定），线程池将创建“非核心线程”临时帮助运行任务。您设置的大于 corePoolSize 参数小于 maximumPoolSize 参数的部分，就是线程池可以临时创建的“非核心线程”的最大数量。这种情况下如果某个线程没有运行任何任务，在等待 keepAliveTime 时间后，这个线程将会被销毁，直到线程池的线程数量重新达到corePoolSize。也就是说，并不是所谓的“非核心线程”才会被回收；而是谁的空闲时间达到 keepAliveTime 这个阀值，就会被回收。直到线程池中线程数量等于 corePoolSize 为止。
  • maximumPoolSize 参数也是当前线程池允许创建的最大线程数量。那么如果您设置的 corePoolSize 参数和您设置的 maximumPoolSize 参数一致时，线程池在任何情况下都不会回收空闲线程。 keepAliveTime 和 timeUnit 也就失去了意义。
  • keepAliveTime 参数和 timeUnit 参数也是配合使用的。 keepAliveTime 参数指明等待时间的量化值， timeUnit 指明量化值单位。例如 keepAliveTime=1，timeUnit 为 TimeUnit.MINUTES ，代表空闲线程的回收阀值为1分钟。
线程池种类

  • CachedThreadPool：一个任务创建一个线程；

    1
    2
    3
    4
    5
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

  • FixedThreadPool：所有任务只能使用固定大小的线程；

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

  • SingleThreadExecutor：相当于大小为 1 的 FixedThreadPool。

    1
    2
    3
    4
    5
    6
    7
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }
完整的任务处理过程

　　1、首先您可以通过线程池提供的 submit() 方法或者 execute() 方法，要求线程池执行某个任务。线程池收到这个要求执行的任务后，会有几种处理情况：

　　　　1.1、如果当前线程池中运行的线程数量还没有达到 corePoolSize 大小时，线程池会创建一个新的线程运行您的任务，无论之前已经创建的线程是否处于空闲状态。

　　　　1.2、如果当前线程池中运行的线程数量已经达到设置的 corePoolSize 大小，线程池会把您的这个任务加入到等待队列中。直到某一个的线程空闲了，线程池会根据您设置的等待队列规则，从队列中取出一个新的任务执行。

　　　　1.3、如果根据队列规则，这个任务无法加入等待队列。这时线程池就会创建一个“非核心线程”直接运行这个任务。注意，如果这种情况下任务执行成功，那么当前线程池中的线程数量一定大于 corePoolSize 。

　　　　1.4、如果这个任务，无法被“核心线程”直接执行，又无法加入等待队列，又无法创建“非核心线程”直接执行，且您没有为线程池设置RejectedExecutionHandler。这时线程池会抛出 RejectedExecutionException 异常，即线程池拒绝接受这个任务。（实际上抛出 RejectedExecutionException 异常的操作，是 ThreadPoolExecutor 线程池中一个默认的 RejectedExecutionHandler 实现：AbortPolicy）

　　2、一旦线程池中某个线程完成了任务的执行，它就会试图到任务等待队列中拿去下一个等待任务（所有的等待任务都实现了BlockingQueue接口，按照接口字面上的理解，这是一个可阻塞的队列接口），它会调用等待队列的 poll() 方法，并停留在哪里。

　　3、当线程池中的线程超过您设置的 corePoolSize 参数，说明当前线程池中有所谓的“非核心线程”。那么当某个线程处理完任务后，如果等待 keepAliveTime 时间后仍然没有新的任务分配给它，那么这个线程将会被回收。线程池回收线程时，对所谓的“核心线程”和“非核心线程”是一视同仁的，直到线程池中线程的数量等于您设置的 corePoolSize 参数时，回收过程才会停止，还有回收的前提条件是线程数大于 corePoolSize。

allowCoreThreadTimeOut

　　前文我们讨论到，线程池回收线程只会发生在当前线程池中线程数量大于 corePoolSize 参数的时候；当线程池中线程数量小于等于 corePoolSize 参数的时候，回收过程就会停止。 allowCoreThreadTimeOut 设置项可以要求线程池：将包括“核心线程”在内的，没有任务分配的任何线程，在等待 keepAliveTime 时间后全部进行回收：

1
2
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(1));
poolExecutor.allowCoreThreadTimeOut(true);
prestartAllCoreThreads

　　前文我们还讨论到，当线程池中的线程还没有达到您设置的 corePoolSize 参数值的时候，如果有新的任务到来，线程池将创建新的线程运行这个任务，无论之前已经创建的线程是否处于空闲状态。这个描述可以用下面的示意图表示出来：

　　prestartAllCoreThreads设置项，可以在线程池创建，但还没有接收到任何任务的情况下，先行创建符合 corePoolSize 参数值的线程数：

1
2
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(1));
poolExecutor.prestartAllCoreThreads();
使用ThreadFactory

　　线程池最主要的一项工作，就是在满足某些条件的情况下创建线程。而在 ThreadPoolExecutor 线程池中，创建线程的工作交给ThreadFactory来完成。要使用线程池，就必须要指定ThreadFactory。

　　 如果我们使用的构造函数时并没有指定使用的 ThreadFactory ，这个时候 ThreadPoolExecutor 会使用一个默认的 ThreadFactory：DefaultThreadFactory。

1
2
3
4
5
6
7
8
9
10
11
12
13
package test.thread.pool;
import java.util.concurrent.ThreadFactory;
/**
 * 测试自定义的一个线程工厂
 */
public class TestThreadFactory implements ThreadFactory {
    @Override
    public Thread newThread(Runnable r) {
        // do something before new thread created;
        // create new thread ， and return
        return new Thread(r);
    }
}
线程池的等待队列

　　在使用 ThreadPoolExecutor 线程池的时候，需要指定一个实现了 BlockingQueue接口的任务等待队列。在 ThreadPoolExecutor 线程池的 API 文档中，一共推荐了三种等待队列，它们是：SynchronousQueue、LinkedBlockingQueue和ArrayBlockingQueue；但通过观察 BlockingQueue 接口的实现情况，您可以发现，能够直接使用的等待队列还有：PriorityBlockingQueue、LinkedBlockingDeque和LinkedTransferQueue。

有限队列

  • SynchronousQueue： 一种阻塞队列，其中每个 put 必须等待一个 take，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。即这是一个内部没有任何容量的阻塞队列，任何一次插入操作的元素都要等待相对的删除/读取操作，否则进行插入操作的线程就要一直等待，反之亦然。

    1
    2
    3
    4
    5
    SynchronousQueue<Object> queue = new SynchronousQueue<Object>();
    // 不要使用add，因为这个队列内部没有任何容量，所以会抛出异常“IllegalStateException”
    // queue.add(new Object());
    // 操作线程会在这里被阻塞，直到有其他操作线程取走这个对象
    queue.put(new Object());

  • ArrayBlockingQueue：一个由数组支持的有界阻塞队列。此队列按 FIFO （先进先出）原则对元素进行排序。新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    // 我们创建了一个ArrayBlockingQueue，并且设置队列空间为2
    ArrayBlockingQueue<Object> arrayQueue = new ArrayBlockingQueue<Object>(2);
    // 插入第一个对象
    arrayQueue.put(new Object());
    // 插入第二个对象
    arrayQueue.put(new Object());
    // 插入第三个对象时，这个操作线程就会被阻塞。
    arrayQueue.put(new Object());
    // 请不要使用add操作，和SynchronousQueue的add操作一样，它们都使用了AbstractQueue中的add实现
无限队列

  • LinkedBlockingQueue： 在 ThreadPoolExecutor 线程池中常应用的等待队列。它可以指定容量也可以不指定容量。由于它具有“无限容量”的特性，所以我还是将它归入了无限队列的范畴（实际上任何无限容量的队列/栈都是有容量的，这个容量就是 Integer.MAX_VALUE ）。 LinkedBlockingQueue 的实现是基于链表结构，而不是类似 ArrayBlockingQueue 那样的数组。但实际使用过程中，您不需要关心它的内部实现，如果您指定了 LinkedBlockingQueue 的容量大小，那么它反映出来的使用特性就和 ArrayBlockingQueue 类似了。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    LinkedBlockingQueue<Object> linkedQueue = new LinkedBlockingQueue<Object>(2);
    linkedQueue.put(new Object());
    // 插入第二个对象
    linkedQueue.put(new Object());
    // 插入第三个对象时，这个操作线程就会被阻塞。
    linkedQueue.put(new Object());
    =======================================
    // 或者如下使用：
    LinkedBlockingQueue<Object> linkedQueue = new LinkedBlockingQueue<Object>();
    linkedQueue.put(new Object());
    // 插入第二个对象
    linkedQueue.put(new Object());
    // 插入第N个对象时，都不会阻塞
    linkedQueue.put(new Object());

  • LinkedBlockingDeque：一个基于链表的双端队列。 LinkedBlockingQueue 的内部结构决定了它只能从队列尾部插入，从队列头部取出元素；但是 LinkedBlockingDeque 既可以从尾部插入/取出元素，还可以从头部插入元素/取出元素。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
LinkedBlockingDeque<TempObject> linkedDeque = new LinkedBlockingDeque<TempObject>();
// push ，可以从队列的头部插入元素
linkedDeque.push(new TempObject(1));
linkedDeque.push(new TempObject(2));
linkedDeque.push(new TempObject(3));
// poll ， 可以从队列的头部取出元素
TempObject tempObject = linkedDeque.poll();
// 这里会打印 tempObject.index = 3
System.out.println("tempObject.index = " + tempObject.getIndex());

// put ， 可以从队列的尾部插入元素
linkedDeque.put(new TempObject(4));
linkedDeque.put(new TempObject(5));
// pollLast , 可以从队列尾部取出元素
tempObject = linkedDeque.pollLast();
// 这里会打印 tempObject.index = 5
System.out.println("tempObject.index = " + tempObject.getIndex());

  • PriorityBlockingQueue： 一个按照优先级进行内部元素排序的无限队列。存放在 PriorityBlockingQueue 中的元素必须实现 Comparable 接口，这样才能通过实现 compareTo() 方法进行排序。优先级最高的元素将始终排在队列的头部； PriorityBlockingQueue 不会保证优先级一样的元素的排序，也不保证当前队列中除了优先级最高的元素以外的元素，随时处于正确排序的位置。

  • LinkedTransferQueue：一个无限队列，它除了具有一般队列的操作特性外（先进先出），还具有一个阻塞特性： LinkedTransferQueue 可以由一对生产者/消费者线程进行操作，当消费者将一个新的元素插入队列后，消费者线程将会一直等待，直到某一个消费者线程将这个元素取走，反之亦然。

　　拒绝任务

　　在 ThreadPoolExecutor 线程池中还有一个重要的接口：RejectedExecutionHandler。当提交给线程池的某一个新任务无法直接被线程池中“核心线程”直接处理，又无法加入等待队列，也无法创建新的线程执行；又或者线程池已经调用 shutdown() 方法停止了工作；又或者线程池不是处于正常的工作状态；这时候 ThreadPoolExecutor 线程池会拒绝处理这个任务，触发您创建 ThreadPoolExecutor 线程池时定义的 RejectedExecutionHandler 接口的实现。

　　ThreadPoolExecutor线程池在创建时，会使用一个默认的 RejectedExecutionHandler 接口实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public class ThreadPoolExecutor extends AbstractExecutorService {

    ......

    /**
     * The default rejected execution handler
     */
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();

    ......

    // 可以看到，ThreadPoolExecutor中的两个没有指定RejectedExecutionHandler
    // 接口的构造函数，都是使用了一个RejectedExecutionHandler接口的默认实现：AbortPolicy
    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) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

    ......
}

　　实际上，在 ThreadPoolExecutor 中已经提供了四种可以直接使用的 RejectedExecutionHandler 接口的实现：

  • CallerRunsPolicy：这个拒绝处理器，将直接运行这个任务的 run 方法。但是，请注意并不是在 ThreadPoolExecutor 线程池中的线程中运行，而是直接调用这个任务实现的 run 方法。
  • AbortPolicy：这个处理器，在任务被拒绝后会创建一个 RejectedExecutionException 异常并抛出。这个处理过程也是 ThreadPoolExecutor 线程池默认的 RejectedExecutionHandler 实现。
  • DiscardPolicy ： DiscardPolicy处理器，将会默默丢弃这个被拒绝的任务，不会抛出异常，也不会通过其他方式执行这个任务的任何一个方法，更不会出现任何的日志提示。
  • DiscardOldestPolicy：这个处理器很有意思。它会检查当前ThreadPoolExecutor线程池的等待队列。并调用队列的 poll() 方法，将当前处于等待队列列头的等待任务强行取出，然后再试图将当前被拒绝的任务提交到线程池执行。

　 CallerRunsPolicy 在非线程池以外直接调用任务的 run 方法，可能会造成线程安全上的问题；

​ DiscardPolicy 默默的忽略掉被拒绝任务，也没有输出日志或者提示，开发人员不会知道线程池的处理过程出现了错误；

​ DiscardOldestPolicy 中 e.getQueue().poll() 的方式好像是科学的，但是如果等待队列出现了容量问题，大多数情况下就是这个线程池的代码出现了 BUG 。

​ 最科学的的还是 AbortPolicy 提供的处理方式：抛出异常，由开发人员进行处理。

-------------本文结束感谢您的阅读-------------