java.util.concurrent概览

java.util.concurrent 包含许多线程安全、测试良好、高性能的并发构建块。不客气地说,创建 java.util.concurrent 的目的就是要实现 Collection 框架对数据结构所执行的并发操作。通过提供一组可靠的、高性能并发构建块,开发人员可以提高并发类的线程安全、可伸缩性、性能、可读性和可靠性。

如果一些类名看起来相似,可能是因为 java.util.concurrent 中的许多概念源自 Doug Lea 的 util.concurrent 库(请参阅 参考资料)。

JDK 5.0 中的并发改进可以分为三组:

1. JVM 级别更改。

    大多数现代处理器对并发对某一硬件级别提供支持,通常以 compare-and-swap (CAS)指令形式。CAS 是一种低级别的、细粒度的技术,它允许多个线程更新一个内存位置,同时能够检测其他线程的冲突并进行恢复。它是许多高性能并发算法的基础。在 JDK 5.0 之前,Java 语言中用于协调线程之间的访问的惟一原语是同步,同步是更重量级和粗粒度的。公开 CAS 可以开发高度可伸缩的并发 Java 类。这些更改主要由 JDK 库类使用,而不是由开发人员使用。

2. 低级实用程序类 -- 锁定和原子类。

    使用 CAS 作为并发原语,ReentrantLock 类提供与 synchronized 原语相同的锁定和内存语义,然而这样可以更好地控制锁定(如计时的锁定等待、锁定轮询和可中断的锁定等待)和提供更好的可伸缩性(竞争时的高性能)。大多数开发人员将不再直接使用 ReentrantLock 类,而是使用在 ReentrantLock 类上构建的高级类。

3. 高级实用程序类。

    这些类实现并发构建块,每个计算机科学文本中都会讲述这些类 -- 信号、互斥、闩锁、屏障、交换程序、线程池和线程安全集合类等。大部分开发人员都可以在应用程序中用这些类,来替换许多(如果不是全部)同步、wait() 和 notify() 的使用,从而提高性能、可读性和正确性。

本文主要内容

本教程将重点介绍 java.util.concurrent 包提供的高级实用程序类 -- 线程安全集合、线程池和同步实用程序。这些是初学者和专家都可以使用的"现成"类。

在第一小节中,我们将回顾并发的基本知识,尽管它不应取代对线程和线程安全的了解。那些一点都不熟悉线程的读者应该先参考一些关于线程的介绍,如"Introduction to Java Threads"教程(请参阅参考资料)。

接下来的几个小节将研究 java.util.concurrent 中的高级实用程序类 -- 线程安全集合、线程池、信号和同步工具。

最后一小节将介绍 java.util.concurrent 中的低级并发构建块,并提供一些性能测评来显示新 java.util.concurrent 类的可伸缩性的改进。

基础知识

  1. 什么是线程?

    所有重要的操作系统都支持进程的概念 – 独立运行的程序,在某种程度上相互隔离。

    线程有时称为 轻量级进程。与进程一样,它们拥有通过程序运行的独立的并发路径,并且每个线程都有自己的程序计数器,称为堆栈和本地变量。然而,线程存在于进程中,它们与同一进程内的其他线程共享内存、文件句柄以及每进程状态。

    今天,几乎每个操作系统都支持线程,允许执行多个可独立调度的线程,以便共存于一个进程中。因为一个进程中的线程是在同一个地址空间中执行的,所以多个线程可以同时访问相同对象,并且它们从同一堆栈中分配对象。虽然这使线程更易于与其他线程共享信息,但也意味着您必须确保线程之间不相互干涉。

    正确使用线程时,线程能带来诸多好处,其中包括更好的资源利用、简化开发、高吞吐量、更易响应的用户界面以及能执行异步处理。

    Java 语言包括用于协调线程行为的原语,从而可以在不违反设计原型或者不破坏数据结构的前提下安全地访问和修改共享变量。

  2. 线程有哪些功能?

    在 Java 程序中存在很多理由使用线程,并且不管开发人员知道线程与否,几乎每个 Java 应用程序都使用线程。许多 J2SE 和 J2EE 工具可以创建线程,如 RMI、Servlet、Enterprise JavaBeans 组件和 Swing GUI 工具包。

  3. 使用线程的理由包括:

    • 更易响应的用户界面。 事件驱动的 GUI 工具包(如 AWT 或 Swing)使用单独的事件线程来处理 GUI 事件。从事件线程中调用通过 GUI 对象注册的事件监听器。然而,如果事件监听器将执行冗长的任务(如文档拼写检查),那么 UI 将出现冻结,因为事件线程直到冗长任务完毕之后才能处理其他事件。通过在单独线程中执行冗长操作,当执行冗长后台任务时,UI 能继续响应。

    • 使用多处理器。 多处理器(MP)系统变得越来越便宜,并且分布越来越广泛。因为调度的基本单位通常是线程,所以不管有多少处理器可用,一个线程的应用程序一次只能在一个处理器上运行。在设计良好的程序中,通过更好地利用可用的计算机资源,多线程能够提高吞吐量和性能。

    • 简化建模。 有效使用线程能够使程序编写变得更简单,并易于维护。通过合理使用线程,个别类可以避免一些调度的详细、交叉存取操作、异步 IO 和资源等待以及其他复杂问题。相反,它们能专注于域的要求,简化开发并改进可靠性。

    • 异步或后台处理。 服务器应用程序可以同时服务于许多远程客户机。如果应用程序从 socket 中读取数据,并且没有数据可以读取,那么对 read() 的调用将被阻塞,直到有数据可读。在单线程应用程序中,这意味着当某一个线程被阻塞时,不仅处理相应请求要延迟,而且处理所有请求也将延迟。然而,如果每个 socket 都有自己的 IO 线程,那么当一个线程被阻塞时,对其他并发请求行为没有影响。

  4. 线程安全

    如果将这些类用于多线程环境中,虽然确保这些类的线程安全比较困难,但线程安全却是必需的。java.util.concurrent 规范进程的一个目标就是提供一组线程安全的、高性能的并发构建块,从而使开发人员能够减轻一些编写线程安全类的负担。

    线程安全类非常难以明确定义,大多数定义似乎都是完全循环的。快速 Google 搜索会显示下列线程安全代码定义的例子,但这些定义(或者更确切地说是描述)通常没什么帮助:

    • . . . can be called from multiple programming threads without unwanted interaction between the threads.

    • . . . may be called by more than on thread at a time without requiring any other action on the caller’s part.

    通过类似这样的定义,不奇怪我们为什么对线程安全如此迷惑。这些定义几乎就是在说”如果可以从多个线程安全调用类,那么该类就是线程安全的”。这当然是线程安全的解释,但对我们区别线程安全类和不安全类没有什么帮助。我们使用”安全”是为了说明什么?

    要成为线程安全的类,首先它必须在单线程环境中正确运行。如果正确实现了类,那么说明它符合规范,对该类的对象的任何顺序的操作(公共字段的读写、公共方法的调用)都不应该使对象处于无效状态;观察将处于无效状态的对象;或违反类的任何变量、前置条件或后置条件。

    而且,要成为线程安全的类,在从多个线程访问时,它必须继续正确运行,而不管运行时环境执行那些线程的调度和交叉,且无需对部分调用代码执行任何其他同步。结果是对线程安全对象的操作将用于按固定的整体一致顺序出现所有线程。

    如果没有线程之间的某种明确协调,比如锁定,运行时可以随意在需要时在多线程中交叉操作执行。

    在 JDK 5.0 之前,确保线程安全的主要机制是 synchronized 原语。访问共享变量(那些可以由多个线程访问的变量)的线程必须使用同步来协调对共享变量的读写访问。java.util.concurrent 包提供了一些备用并发原语,以及一组不需要任何其他同步的线程安全实用程序类。

  5. 令人厌烦的并发

    即使您的程序从没有明确创建线程,也可能会有许多工具或框架代表您创建了线程,这时要求从这些线程调用的类是线程安全的。这样会对开发人员带来较大的设计和实现负担,因为开发线程安全类比开发非线程安全类有更多要注意的事项,且需要更多的分析。

  6. AWT 和 Swing

    这些 GUI 工具包创建了称为时间线程的后台线程,将从该线程调用通过 GUI 组件注册的监听器。因此,实现这些监听器的类必须是线程安全的。

  7. TimerTask

    JDK 1.3 中引入的 TimerTask 工具允许稍后执行任务或计划定期执行任务。在 Timer 线程中执行 TimerTask 事件,这意味着作为 TimerTask 执行的任务必须是线程安全的。

  8. Servlet 和 JavaServer Page 技术

    Servlet 容器可以创建多个线程,在多个线程中同时调用给定 servlet,从而进行多个请求。因此 servlet 类必须是线程安全的。

  9. RMI

    远程方法调用(remote method invocation,RMI)工具允许调用其他 JVM 中运行的操作。实现远程对象最普遍的方法是扩展 UnicastRemoteObject。例示 UnicastRemoteObject 时,它是通过 RMI 调度器注册的,该调度器可能创建一个或多个线程,将在这些线程中执行远程方法。因此,远程类必须是线程安全的。

    正如所看到的,即使应用程序没有明确创建线程,也会发生许多可能会从其他线程调用类的情况。幸运的是,java.util.concurrent 中的类可以大大简化编写线程安全类的任务。

例子 -- 非线程安全 servlet

下列 servlet 看起来像无害的留言板 servlet,它保存每个来访者的姓名。然而,该 servlet 不是线程安全的,而这个 servlet 应该是线程安全的。问题在于它使用 HashSet 存储来访者的姓名,HashSet 不是线程安全的类。

当我们说这个 servlet 不是线程安全的时,是说它所造成的破坏不仅仅是丢失留言板输入。在最坏的情况下,留言板数据结构都可能被破坏并且无法恢复。

public class UnsafeGuestbookServlet extends HttpServlet {

private Set visitorSet = new HashSet();

protected void doGet(HttpServletRequest httpServletRequest,

HttpServletResponse httpServletResponse) throws ServletException, IOException {

String visitorName = httpServletRequest.getParameter("NAME");

if (visitorName != null)

visitorSet.add(visitorName);

}

}

通过将 visitorSet 的定义更改为下列代码,可以使该类变为线程安全的:

private Set visitorSet = Collections.synchronizedSet(new HashSet());

如上所示的例子显示线程的内置支持是一把双刃剑 -- 虽然它使构建多线程应用程序变得很容易,但它同时要求开发人员更加注意并发问题,甚至在使用留言板 servlet 这样普通的东西时也是如此。

线程安全集合

JDK 1.2 中引入的 Collection 框架是一种表示对象集合的高度灵活的框架,它使用基本接口 List、Set 和 Map。通过 JDK 提供每个集合的多次实现(HashMap、Hashtable、TreeMap、WeakHashMap、HashSet、TreeSet、Vector、ArrayList、LinkedList 等等)。其中一些集合已经是线程安全的(Hashtable 和 Vector),通过同步的封装工厂(Collections.synchronizedMap()、synchronizedList() 和 synchronizedSet()),其余的集合均可表现为线程安全的。
java.util.concurrent 包添加了多个新的线程安全集合类(ConcurrentHashMap、CopyOnWriteArrayList 和 CopyOnWriteArraySet)。这些类的目的是提供高性能、高度可伸缩性、线程安全的基本集合类型版本。
java.util 中的线程集合仍有一些缺点。例如,在迭代锁定时,通常需要将该锁定保留在集合中,否则,会有抛出 ConcurrentModificationException 的危险。(这个特性有时称为条件线程安全;有关的更多说明,请参阅参考资料。)此外,如果从多个线程频繁地访问集合,则常常不能很好地执行这些类。java.util.concurrent 中的新集合类允许通过在语义中的少量更改来获得更高的并发。
JDK 5.0 还提供了两个新集合接口 – Queue 和 BlockingQueue。Queue 接口与 List 类似,但它只允许从后面插入,从前面删除。通过消除 List 的随机访问要求,可以创建比现有 ArrayList 和 LinkedList 实现性能更好的 Queue 实现。因为 List 的许多应用程序实际上不需要随机访问,所以Queue 通常可以替代 List,来获得更好的性能。

  1. 弱一致的迭代器

    java.util 包中的集合类都返回 fail-fast 迭代器,这意味着它们假设线程在集合内容中进行迭代时,集合不会更改它的内容。如果 fail-fast 迭代器检测到在迭代过程中进行了更改操作,那么它会抛出 ConcurrentModificationException,这是不可控异常。

    在迭代过程中不更改集合的要求通常会对许多并发应用程序造成不便。相反,比较好的是它允许并发修改并确保迭代器只要进行合理操作,就可以提供集合的一致视图,如 java.util.concurrent 集合类中的迭代器所做的那样。

    java.util.concurrent 集合返回的迭代器称为弱一致的(weakly consistent)迭代器。对于这些类,如果元素自从迭代开始已经删除,且尚未由 next() 方法返回,那么它将不返回到调用者。如果元素自迭代开始已经添加,那么它可能返回调用者,也可能不返回。在一次迭代中,无论如何更改底层集合,元素不会被返回两次。

  2. CopyOnWriteArrayList 和 CopyOnWriteArraySet

    可以用两种方法创建线程安全支持数据的 List – Vector 或封装 ArrayList 和 Collections.synchronizedList()。java.util.concurrent 包添加了名称繁琐的 CopyOnWriteArrayList。为什么我们想要新的线程安全的List类?为什么Vector还不够?

    最简单的答案是与迭代和并发修改之间的交互有关。使用 Vector 或使用同步的 List 封装器,返回的迭代器是 fail-fast 的,这意味着如果在迭代过程中任何其他线程修改 List,迭代可能失败。

  3. Vector 的非常普遍的应用程序是存储通过组件注册的监听器的列表。当发生适合的事件时,该组件将在监听器的列表中迭代,调用每个监听器。为了防止 ConcurrentModificationException,迭代线程必须复制列表或锁定列表,以便进行整体迭代,而这两种情况都需要大量的性能成本。

  4. CopyOnWriteArrayList 类通过每次添加或删除元素时创建支持数组的新副本,避免了这个问题,但是进行中的迭代保持对创建迭代器时的当前副本进行操作。虽然复制也会有一些成本,但是在许多情况下,迭代要比修改多得多,在这些情况下,写入时复制要比其他备用方法具有更好的性能和并发性。

  5. 如果应用程序需要 Set 语义,而不是 List,那么还有一个 Set 版本 – CopyOnWriteArraySet。

  6. ConcurrentHashMap

    正如已经存在线程安全的 List 的实现,您可以用多种方法创建线程安全的、基于 hash 的 Map – Hashtable,并使用 Collections.synchronizedMap() 封装 HashMap。JDK 5.0 添加了 ConcurrentHashMap 实现,该实现提供了相同的基本线程安全的 Map 功能,但它大大提高了并发性。

    Hashtable 和 synchronizedMap 所采取的获得同步的简单方法(同步 Hashtable 中或者同步的 Map 封装器对象中的每个方法)有两个主要的不足。首先,这种方法对于可伸缩性是一种障碍,因为一次只能有一个线程可以访问 hash 表。同时,这样仍不足以提供真正的线程安全性,许多公用的混合操作仍然需要额外的同步。虽然诸如 get() 和 put() 之类的简单操作可以在不需要额外同步的情况下安全地完成,但还是有一些公用的操作序列,例如迭代或者 put-if-absent(空则放入),需要外部的同步,以避免数据争用。

    Hashtable 和 Collections.synchronizedMap 通过同步每个方法获得线程安全。这意味着当一个线程执行一个 Map 方法时,无论其他线程要对 Map 进行什么样操作,都不能执行,直到第一个线程结束才可以。

    对比来说,ConcurrentHashMap 允许多个读取几乎总是并发执行,读和写操作通常并发执行,多个同时写入经常并发执行。结果是当多个线程需要访问同一 Map 时,可以获得更高的并发性。

    在大多数情况下,ConcurrentHashMap 是 Hashtable或 Collections.synchronizedMap(new HashMap()) 的简单替换。然而,其中有一个显著不同,即 ConcurrentHashMap 实例中的同步不锁定映射进行独占使用。实际上,没有办法锁定 ConcurrentHashMap 进行独占使用,它被设计用于进行并发访问。为了使集合不被锁定进行独占使用,还提供了公用的混合操作的其他(原子)方法,如 put-if-absent。ConcurrentHashMap 返回的迭代器是弱一致的,意味着它们将不抛出ConcurrentModificationException ,将进行”合理操作”来反映迭代过程中其他线程对 Map 的修改。

  7. 队列

    原始集合框架包含三个接口:List、Map 和 Set。List 描述了元素的有序集合,支持完全随即访问 – 可以在任何位置添加、提取或删除元素。

    LinkedList 类经常用于存储工作元素(等待执行的任务)的列表或队列。然而,List 提供的灵活性比该公用应用程序所需要的多得多,这个应用程序通常在后面插入元素,从前面删除元素。但是要支持完整 List 接口则意味着 LinkedList 对于这项任务不像原来那样有效。Queue 接口比 List 简单得多,仅包含 put() 和 take() 方法,并允许比 LinkedList 更有效的实现。

    Queue 接口还允许实现来确定存储元素的顺序。ConcurrentLinkedQueue 类实现先进先出(first-in-first-out,FIFO)队列,而 PriorityQueue 类实现优先级队列(也称为堆),它对于构建调度器非常有用,调度器必须按优先级或预期的执行时间执行任务。

    interface Queue extends Collection {

    boolean offer(E x);

    E poll();

    E remove() throws NoSuchElementException;

    E peek();

    E element() throws NoSuchElementException;

    }

    实现 Queue 的类是:

     • LinkedList 已经进行了改进来实现 Queue。
    
     • PriorityQueue 非线程安全的优先级对列(堆)实现,根据自然顺序或比较器返回元素。
    
     • ConcurrentLinkedQueue 快速、线程安全的、无阻塞 FIFO 队列。
  8. 任务管理之线程创建

    线程最普遍的一个应用程序是创建一个或多个线程,以执行特定类型的任务。Timer 类创建线程来执行 TimerTask 对象,Swing 创建线程来处理 UI 事件。在这两种情况中,在单独线程中执行的任务都假定是短期的,这些线程是为了处理大量短期任务而存在的。

    在其中每种情况中,这些线程一般都有非常简单的结构:

    while (true) {

    if (no tasks)

    wait for a task;

    execute the task;

    }

    通过例示从 Thread 获得的对象并调用 Thread.start() 方法来创建线程。可以用两种方法创建线程:通过扩展 Thread 和覆盖 run() 方法,或者通过实现 Runnable 接口和使用 Thread(Runnable) 构造函数:

    class WorkerThread extends Thread {

    public void run() { /* do work */ }

    }

    Thread t = new WorkerThread();

    t.start();

    或者:

    Thread t = new Thread(new Runnable() {

    public void run() { /* do work */ }

    }

    t.start();

  9. 重新使用线程

    因为多个原因,类似 Swing GUI 的框架为事件任务创建单一线程,而不是为每项任务创建新的线程。首先是因为创建线程会有间接成本,所以创建线程来执行简单任务将是一种资源浪费。通过重新使用事件线程来处理多个事件,启动和拆卸成本(随平台而变)会分摊在多个事件上。

    Swing 为事件使用单一后台线程的另一个原因是确保事件不会互相干涉,因为直到前一事件结束,下一事件才开始处理。该方法简化了事件处理程序的编写。

    使用多个线程,将要做更多的工作来确保一次仅一个线程地执行线程相关的代码。

  10. 如何不对任务进行管理

    大多数服务器应用程序(如 Web 服务器、POP 服务器、数据库服务器或文件服务器)代表远程客户机处理请求,这些客户机通常使用 socket 连接到服务器。对于每个请求,通常要进行少量处理(获得该文件的代码块,并将其发送回 socket),但是可能会有大量(且不受限制)的客户机请求服务。

    用于构建服务器应用程序的简单化模型会为每个请求创建新的线程。下列代码段实现简单的 Web 服务器,它接受端口 80 的 socket 连接,并创建新的线程来处理请求。不幸的是,该代码不是实现 Web 服务器的好方法,因为在重负载条件下它将失败,停止整台服务器。

    class UnreliableWebServer {

    public static void main(String[] args) {

    ServerSocket socket = new ServerSocket(80);

    while (true) {

    final Socket connection = socket.accept();

    Runnable r = new Runnable() {

    public void run() {

    handleRequest(connection);

    }

    };

    // Don't do this!

    new Thread(r).start();

    }

    }

    }

    当服务器被请求吞没时,UnreliableWebServer 类不能很好地处理这种情况。每次有请求时,就会创建新的类。根据操作系统和可用内存,可以创建的线程数是有限的。

    不幸的是,您通常不知道限制是多少 – 只有当应用程序因为 OutOfMemoryError 而崩溃时才发现。

    如果足够快地在这台服务器上抛出请求的话,最终其中一个线程创建将失败,生成的 Error 会关闭整个应用程序。当一次仅能有效支持很少线程时,没有必要创建上千个

    线程,无论如何,这样使用资源可能会损害性能。创建线程会使用相当一部分内存,其中包括有两个堆栈(Java 和 C),以及每线程数据结构。如果创建过多线程,其中

    每个线程都将占用一些 CPU 时间,结果将使用许多内存来支持大量线程,每个线程都运行得很慢。这样就无法很好地使用计算资源。

  11. 使用线程池解决问题

    为任务创建新的线程并不一定不好,但是如果创建任务的频率高,而平均任务持续时间低,我们可以看到每项任务创建一个新的线程将产生性能(如果负载不可预知,还有稳定性)问题。

    如果不是每项任务创建一个新的线程,则服务器应用程序必须采取一些方法来限制一次可以处理的请求数。这意味着每次需要启动新的任务时,它不能仅调用下列代码。

    new Thread(runnable).start()

    管理一大组小任务的标准机制是组合工作队列和线程池。工作队列就是要处理的任务的队列,前面描述的 Queue 类完全适合。线程池是线程的集合,每个线程都提取公用工作队列。当一个工作线程完成任务处理后,它会返回队列,查看是否有其他任务需要处理。如果有,它会转移到下一个任务,并开始处理。

    线程池为线程生命周期间接成本问题和资源崩溃问题提供了解决方案。通过对多个任务重新使用线程,创建线程的间接成本将分布到多个任务中。作为一种额外好处,因为请求到达时,线程已经存在,从而可以消除由创建线程引起的延迟。因此,可以立即处理请求,使应用程序更易响应。而且,通过正确调整线程池中的线程数,可以强制超出特定限制的任何请求等待,直到有线程可以处理它,它们等待时所消耗的资源要少于使用额外线程所消耗的资源,这样可以防止资源崩溃。

  12. Executor 框架

    java.util.concurrent 包中包含灵活的线程池实现,但是更重要的是,它包含用于管理实现 Runnable 的任务的执行的整个框架。该框架称为 Executor 框架。

    Executor 接口相当简单。它描述将运行 Runnable 的对象:

    public interface Executor {

    void execute(Runnable command);

    }

    任务运行于哪个线程不是由该接口指定的,这取决于使用的 Executor 的实现。它可以运行于后台线程,如 Swing 事件线程,或者运行于线程池,或者调用线程,或者新的线程,它甚至可以运行于其他 JVM!通过同步的 Executor 接口提交任务,从任务执行策略中删除任务提交。Executor 接口独自关注任务提交 – 这是Executor 实现的选择,确定执行策略。这使在部署时调整执行策略(队列限制、池大小、优先级排列等等)更加容易,更改的代码最少。

    java.util.concurrent 中的大多数 Executor 实现还实现 ExecutorService 接口,这是对 Executor 的扩展,它还管理执行服务的生命周期。这使它们更易于管理,并向生命可能比单独 Executor 的生命更长的应用程序提供服务。

    public interface ExecutorService extends Executor {

    void shutdown();

    List shutdownNow();

    boolean isShutdown();

    boolean isTerminated();

    boolean awaitTermination(long timeout,

    TimeUnit unit);

    // other convenience methods for submitting tasks

    }
  13. Executor

    java.util.concurrent 包包含多个 Executor 实现,每个实现都实现不同的执行策略。什么是执行策略?执行策略定义何时在哪个线程中运行任务,执行任务可能消耗的资源级别(线程、内存等等),以及如果执行程序超载该怎么办。

    执行程序通常通过工厂方法例示,而不是通过构造函数。Executors 类包含用于构造许多不同类型的 Executor 实现的静态工厂方法:

    • Executors.newCachedThreadPool() 创建不限制大小的线程池,但是当以前创建的线程可以使用时将重新使用那些线程。如果没有现有线程可用,

    • 将创建新的线程并将其添加到池中。使用不到 60 秒的线程将终止并从缓存中删除。

    • Executors.newFixedThreadPool(int n) 创建线程池,其重新使用在不受限制的队列之外运行的固定线程组。在关闭前,所有线程都会因为执行

    • 过程中的失败而终止,如果需要执行后续任务,将会有新的线程来代替这些线程。

    • Executors.newSingleThreadExecutor() 创建 Executor,其使用在不受限制的队列之外运行的单一工作线程,与 Swing 事件线程非常相似。

    • 保证顺序执行任务,在任何给定时间,不会有多个任务处于活动状态。

    更可靠的 Web 服务器 – 使用 Executor

    前面 如何不对任务进行管理 中的代码显示了如何不用编写可靠服务器应用程序。幸运的是,修复这个示例非常简单,只需将 Thread.start() 调用替换为向 Executor 提交任务即可:

    class ReliableWebServer {

    Executor pool =

    Executors.newFixedThreadPool(7);

    public static void main(String[] args) {

    ServerSocket socket = new ServerSocket(80);

    while (true) {

    final Socket connection = socket.accept();

    Runnable r = new Runnable() {

    public void run() {

    handleRequest(connection);

    }

    };

    pool.execute(r);

    }

    }

    }

    注意,本例与前例之间的区别仅在于 Executor 的创建以及如何提交执行的任务。

  1. 定制 ThreadPoolExecutor

Executors 中的 newFixedThreadPool 和 newCachedThreadPool 工厂方法返回的 Executor 是类 ThreadPoolExecutor 的实例,是高度可定制的。
通过使用包含 ThreadFactory 变量的工厂方法或构造函数的版本,可以定义池线程的创建。ThreadFactory 是工厂对象,其构造执行程序要使用的新线程。
使用定制的线程工厂,创建的线程可以包含有用的线程名称,并且这些线程是守护线程,属于特定线程组或具有特定优先级。
下面是线程工厂的例子,它创建守护线程,而不是创建用户线程:

public class DaemonThreadFactory implements ThreadFactory {

public Thread newThread(Runnable r) {

Thread thread = new Thread(r);

thread.setDaemon(true);

return thread;

}

}

有时,Executor 不能执行任务,因为它已经关闭或者因为 Executor 使用受限制队列存储等待任务,而该队列已满。在这种情况下,需要咨询执行程序的 RejectedExecutionHandler 来确定如何处理任务 -- 抛出异常(默认情况),放弃任务,在调用者的线程中执行任务,或放弃队列中最早的任务以为新任务腾出空间。ThreadPoolExecutor.setRejectedExecutionHandler 可以设置拒绝的执行处理程序。 还可以扩展 ThreadPoolExecutor,并覆盖方法 beforeExecute 和 afterExecute,以添加装置,添加记录,添加计时,重新初始化线程本地变量,或进行其他执行定制。 需要特别考虑的问题 使用 Executor 框架会从执行策略中删除任务提交,一般情况下,人们希望这样,那是因为它允许我们灵活地调整执行策略,不必更改许多位置的代码。然而,当提交代码暗含假设特定执行策略时,存在多种情况,在这些情况下,重要的是选择的 Executor 实现一致的执行策略。 这类情况中的其中的一种就是一些任务同时等待其他任务完成。在这种情况下,当线程池没有足够的线程时,如果所有当前执行的任务都在等待另一项任务,而该任务因为线程池已满不能执行,那么线程池可能会死锁。 另一种相似的情况是一组线程必须作为共同操作组一起工作。在这种情况下,需要确保线程池能够容纳所有线程。 如果应用程序对特定执行程序进行了特定假设,那么应该在 Executor 定义和初始化的附近对这些进行说明,从而使善意的更改不会破坏应用程序的正确功能。 调整线程池 创建 Executor 时,人们普遍会问的一个问题是"线程池应该有多大?"。当然,答案取决于硬件和将执行的任务类型(它们是受计算限制或是受 IO 的限制?)。 如果线程池太小,资源可能不能被充分利用,在一些任务还在工作队列中等待执行时,可能会有处理器处于闲置状态。 另一方面,如果线程池太大,则将有许多有效线程,因为大量线程或有效任务使用内存,或者因为每项任务要比使用少量线程有更多上下文切换,性能可能会受损。 所以假设为了使处理器得到充分使用,线程池应该有多大?如果知道系统有多少处理器和任务的计算时间和等待时间的近似比率,Amdahl 法则提供很好的近似公式。 用 WT 表示每项任务的平均等待时间,ST 表示每项任务的平均服务时间(计算时间)。则 WT/ST 是每项任务等待所用时间的百分比。对于 N 处理器系统,池中可以近似有 N*(1+WT/ST) 个线程。 好的消息是您不必精确估计 WT/ST。"合适的"池大小的范围相当大;只需要避免"过大"和"过小"的极端情况即可。
  1. Future 接口

    Future 接口允许表示已经完成的任务、正在执行过程中的任务或者尚未开始执行的任务。通过 Future 接口,可以尝试取消尚未完成的任务,查询任务已经完成还是取消了,以及提取(或等待)任务的结果值。

    FutureTask 类实现了 Future,并包含一些构造函数,允许将 Runnable 或 Callable(会产生结果的 Runnable)和 Future 接口封装。因为 FutureTask 也实现 Runnable,所以可以只将 FutureTask 提供给 Executor。一些提交方法(如 ExecutorService.submit())除了提交任务之外,还将返回 Future 接口。

    Future.get() 方法检索任务计算的结果(或如果任务完成,但有异常,则抛出 ExecutionException)。如果任务尚未完成,那么 Future.get() 将被阻塞,直到任务完成;如果任务已经完成,那么它将立即返回结果。

    使用 Future 构建缓存

    该示例代码与 java.util.concurrent 中的多个类关联,突出显示了 Future 的功能。它实现缓存,使用 Future 描述缓存值,该值可能已经计算,或者可能在其他线程中"正在构造"。
    
    它利用 ConcurrentHashMap 中的原子 putIfAbsent() 方法,确保仅有一个线程试图计算给定关键字的值。如果其他线程随后请求同一关键字的值,它仅能等待(通过 Future.get() 的帮助)第一个线程完成。因此两个线程不会计算相同的值。
    public class Cache {

    ConcurrentMap> map = new ConcurrentHashMap();

    Executor executor = Executors.newFixedThreadPool(8);

    public V get(final K key) {

    FutureTask f = map.get(key);

    if (f == null) {

    Callable c = new Callable() {

    public V call() {

    // return value associated with key

    }

    };

    f = new FutureTask(c);

    FutureTask old = map.putIfAbsent(key, f);

    if (old == null)

    executor.execute(f);

    else

    f = old;

    }

    return f.get();

    }

    }
  2. CompletionService

    CompletionService 将执行服务与类似 Queue 的接口组合,从任务执行中删除任务结果的处理。CompletionService 接口包含用来提交将要执行的任务的 submit() 方法和用来询问下一完成任务的 take()/poll() 方法。

    CompletionService 允许应用程序结构化,使用 Producer/Consumer 模式,其中生产者创建任务并提交,消费者请求完成任务的结果并处理这些结果。CompletionService 接口由 ExecutorCompletionService 类实现,该类使用 Executor 处理任务并从 CompletionService 导出 submit/poll/take 方法。

    下列代码使用 Executor 和 CompletionService 来启动许多”solver”任务,并使用第一个生成非空结果的任务的结果,然后取消其余任务:

    void solve(Executor e, Collection> solvers)

    throws InterruptedException {

    CompletionService ecs = new ExecutorCompletionService(e);

    int n = solvers.size();

    List> futures = new ArrayList>(n);

    Result result = null;

    try {

    for (Callable s : solvers)

    futures.add(ecs.submit(s));

    for (int i = 0; i < n; ++i) {

    try {

    Result r = ecs.take().get();

    if (r != null) {

    result = r;

    break;

    }

    } catch(ExecutionException ignore) {}

    }

    }

    finally {

    for (Future f : futures)

    f.cancel(true);

    }

    if (result != null)

    use(result);

    }

    java.util.concurrent 中其他类别的有用的类也是同步工具。这组类相互协作,控制一个或多个线程的执行流。

    Semaphore、CyclicBarrier、CountdownLatch 和 Exchanger 类都是同步工具的例子。每个类都有线程可以调用的方法,方法是否被阻塞取决于正在使用的特定同步工具的状态和规则。

  3. Semaphore

    Semaphore 类实现标准 Dijkstra 计数信号。计数信号可以认为具有一定数量的许可权,该许可权可以获得或释放。如果有剩余的许可权,acquire() 方法将成功,否则该方法将被阻塞,直到有可用的许可权(通过其他线程释放许可权)。线程一次可以获得多个许可权。

    计数信号可以用于限制有权对资源进行并发访问的线程数。该方法对于实现资源池或限制 Web 爬虫(Web crawler)中的输出 socket 连接非常有用。

    注意信号不跟踪哪个线程拥有多少许可权;这由应用程序来决定,以确保何时线程释放许可权,该信号表示其他线程拥有许可权或者正在释放许可权,以及其他线程知道它的许可权已释放。

  4. 互斥

    计数信号的一种特殊情况是互斥,或者互斥信号。互斥就是具有单一许可权的计数信号,意味着在给定时间仅一个线程可以具有许可权(也称为二进制信号)。互斥可以用于管理对共享资源的独占访问。

    虽然互斥许多地方与锁定一样,但互斥还有一个锁定通常没有的其他功能,就是互斥可以由具有许可权的线程之外的其他线程来释放。这在死锁恢复时会非常有用。

    CyclicBarrier 类可以帮助同步,它允许一组线程等待整个线程组到达公共屏障点。CyclicBarrier 是使用整型变量构造的,其确定组中的线程数。当一个线程到达屏障时(通过调用 CyclicBarrier.await()),它会被阻塞,直到所有线程都到达屏障,然后在该点允许所有线程继续执行。该操作与许多家庭逛商业街相似 – 每个家庭成员都自己走,并商定 1:00 在电影院集合。当您到电影院但不是所有人都到了时,您会坐下来等其他人到达。然后所有人一起离开。

    认为屏障是循环的是因为它可以重新使用;一旦所有线程都已经在屏障处集合并释放,则可以将该屏障重新初始化到它的初始状态。 还可以指定在屏障处等待时的超时;如果在该时间内其余线程还没有到达屏障,则认为屏障被打破,所有正在等待的线程会收到 BrokenBarrierException。

    下列代码将创建 CyclicBarrier 并启动一组线程,每个线程将计算问题的一部分,等待所有其他线程结束之后,再检查解决方案是否达成一致。如果不一致,那么每个工作线程将开始另一个迭代。该例将使用 CyclicBarrier 变量,它允许注册 Runnable,在所有线程到达屏障但还没有释放任何线程时执行 Runnable。

    class Solver { // Code sketch

    void solve(final Problem p, int nThreads) {

    final CyclicBarrier barrier =

    new CyclicBarrier(nThreads,

    new Runnable() {

    public void run() { p.checkConvergence(); }}

    );

    for (int i = 0; i < nThreads; ++i) {

    final int id = i;

    Runnable worker = new Runnable() {

    final Segment segment = p.createSegment(id);

    public void run() {

    try {

    while (!p.converged()) {

    segment.update();

    barrier.await();

    }

    }

    catch(Exception e) { return; }

    }

    };

    new Thread(worker).start();

    }

    }
  5. CountdownLatch

    CountdownLatch 类与 CyclicBarrier 相似,因为它的角色是对已经在它们中间分摊了问题的一组线程进行协调。它也是使用整型变量构造的,指明计数的初始值,但是与 CyclicBarrier 不同的是,CountdownLatch 不能重新使用。

    其中,CyclicBarrier 是到达屏障的所有线程的大门,只有当所有线程都已经到达屏障或屏障被打破时,才允许这些线程通过,CountdownLatch 将到达和等待功能分离。任何线程都可以通过调用 countDown() 减少当前计数,这种不会阻塞线程,而只是减少计数。await() 方法的行为与 CyclicBarrier.await() 稍微有所不同,调用 await() 任何线程都会被阻塞,直到闩锁计数减少为零,在该点等待的所有线程才被释放,对 await() 的后续调用将立即返回。

    当问题已经分解为许多部分,每个线程都被分配一部分计算时,CountdownLatch 非常有用。在工作线程结束时,它们将减少计数,协调线程可以在闩锁处等待当前这一批计算结束,然后继续移至下一批计算。

    相反地,具有计数 1 的 CountdownLatch 类可以用作”启动大门”,来立即启动一组线程;工作线程可以在闩锁处等待,协调线程减少计数,从而立即释放所有工作线程。下例使用两个 CountdownLatche。一个作为启动大门,一个在所有工作线程结束时释放线程:

    class Driver { // ...

    void main() throws InterruptedException {

    CountDownLatch startSignal = new CountDownLatch(1);

    CountDownLatch doneSignal = new CountDownLatch(N);

    for (int i = 0; i < N; ++i) // create and start threads

    new Thread(new Worker(startSignal, doneSignal)).start();

    doSomethingElse(); // don't let them run yet

    startSignal.countDown(); // let all threads proceed

    doSomethingElse();

    doneSignal.await(); // wait for all to finish

    }

    }

    class Worker implements Runnable {

    private final CountDownLatch startSignal;

    private final CountDownLatch doneSignal;

    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {

    this.startSignal = startSignal;

    this.doneSignal = doneSignal;

    }

    public void run() {

    try {

    startSignal.await();

    doWork();

    doneSignal.countDown();

    } catch (InterruptedException ex) {} // return;

    }

    }
  6. Exchanger 类方便了两个共同操作线程之间的双向交换;这样,就像具有计数为 2 的 CyclicBarrier,并且两个线程在都到达屏障时可以”交换”一些状态。(Exchanger 模式有时也称为聚集。)

    Exchanger 通常用于一个线程填充缓冲(通过读取 socket),而另一个线程清空缓冲(通过处理从 socket 收到的命令)的情况。当两个线程在屏障处集合时,它们交换缓冲。下列代码说明了这项技术:

    class FillAndEmpty {

    Exchanger exchanger = new Exchanger();

    DataBuffer initialEmptyBuffer = new DataBuffer();

    DataBuffer initialFullBuffer = new DataBuffer();

    class FillingLoop implements Runnable {

    public void run() {

    DataBuffer currentBuffer = initialEmptyBuffer;

    try {

    while (currentBuffer != null) {

    addToBuffer(currentBuffer);

    if (currentBuffer.full())

    currentBuffer = exchanger.exchange(currentBuffer);

    }

    } catch (InterruptedException ex) { ... handle ... }

    }

    }

    class EmptyingLoop implements Runnable {

    public void run() {

    DataBuffer currentBuffer = initialFullBuffer;

    try {

    while (currentBuffer != null) {

    takeFromBuffer(currentBuffer);

    if (currentBuffer.empty())

    currentBuffer = exchanger.exchange(currentBuffer);

    }

    } catch (InterruptedException ex) { ... handle ...}

    }

    }

    void start() {

    new Thread(new FillingLoop()).start();

    new Thread(new EmptyingLoop()).start();

    }

    }

  7. 锁定和原子之Lock

    Java 语言内置了锁定工具 – synchronized 关键字。当线程获得监视器时(内置锁定),其他线程如果试图获得相同锁定,那么它们将被阻塞,直到第一个线程释放该锁定。同步还确保随后获得相同锁定的线程可以看到之前的线程在具有该锁定时所修改的变量的值,从而确保如果类正确地同步了共享状态的访问权,那么线程将不会看到变量的”失效”值,这是缓存或编译器优化的结果。

    虽然同步没有什么问题,但它有一些限制,在一些高级应用程序中会造成不便。Lock 接口将内置监视器锁定的锁定行为普遍化,允许多个锁定实现,同时提供一些内置锁定缺少的功能,如计时的等待、可中断的等待、锁定轮询、每个锁定有多个条件等待集合以及无阻塞结构的锁定。

    interface Lock {

    void lock();

    void lockInterruptibly() throws IE;

    boolean tryLock();

    boolean tryLock(long time,

    TimeUnit unit) throws IE;

    void unlock();

    Condition newCondition() throws

    UnsupportedOperationException;

    }
  8. ReentrantLock

    ReentrantLock 是具有与隐式监视器锁定(使用 synchronized 方法和语句访问)相同的基本行为和语义的 Lock 的实现,但它具有扩展的能力。

    作为额外收获,在竞争条件下,ReentrantLock 的实现要比现在的 synchronized 实现更具有可伸缩性。(有可能在 JVM 的将来版本中改进 synchronized 的竞争性能。)

    这意味着当许多线程都竞争相同锁定时,使用 ReentrantLock 的吞吐量通常要比 synchronized 好。换句话说,当许多线程试图访问 ReentrantLock 保护的共享资源时,JVM 将花费较少的时间来调度线程,而用更多个时间执行线程。

    虽然 ReentrantLock 类有许多优点,但是与同步相比,它有一个主要缺点 – 它可能忘记释放锁定。建议当获得和释放 ReentrantLock 时使用下列结构:

    Lock lock = new ReentrantLock();

    ...

    lock.lock();

    try {

    // perform operations protected by lock

    }

    catch(Exception ex) {

    // restore invariants

    }

    finally {

    lock.unlock();

    }

    因为锁定失误(忘记释放锁定)的风险,所以对于基本锁定,强烈建议您继续使用 synchronized,除非真的需要 ReentrantLock 额外的灵活性和可伸缩性。

    ReentrantLock 是用于高级应用程序的高级工具 – 有时需要,但有时用原来的方法就很好。

  9. Condition

    就像 Lock 接口是同步的具体化,Condition 接口是 Object 中 wait() 和 notify() 方法的具体化。Lock 中的一个方法是 newCondition(),它要求锁定向该锁定返回新的 Condition 对象限制。await()、signal() 和 signalAll() 方法类似于 wait()、notify() 和 notifyAll(),但增加了灵活性,每个 Lock 都可以创建多个条件变量。这简化了一些并发算法的实现。

  10. ReadWriteLock

    ReentrantLock 实现的锁定规则非常简单 – 每当一个线程具有锁定时,其他线程必须等待,直到该锁定可用。有时,当对数据结构的读取通常多于修改时,可以使用更复杂的称为读写锁定的锁定结构,它允许有多个并发读者,同时还允许一个写入者独占锁定。该方法在一般情况下(只读)提供了更大的并发性,同时在必要时仍提供独占访问的安全性。ReadWriteLock 接口和 ReentrantReadWriteLock 类提供这种功能 – 多读者、单写入者锁定规则,可以用这种功能来保护共享的易变资源。

  11. 原子变量

    即使大多数用户将很少直接使用它们,原子变量类(AtomicInteger、AtomicLong、AtomicReference 等等)也有充分理由是最显著的新并发类。这些类公开对 JVM 的低级别改进,允许进行具有高度可伸缩性的原子读-修改-写操作。大多数现代 CPU 都有原子读-修改-写的原语,比如比较并交换(CAS)或加载链接/条件存储(LL/SC)。原子变量类使用硬件提供的最快的并发结构来实现。

    许多并发算法都是根据对计数器或数据结构的比较并交换操作来定义的。通过暴露高性能的、高度可伸缩的 CAS 操作(以原子变量的形式),用 Java 语言实现高性能、无等待、无锁定的并发算法已经变得可行。

    几乎 java.util.concurrent 中的所有类都是在 ReentrantLock 之上构建的,ReentrantLock 则是在原子变量类的基础上构建的。所以,虽然仅少数并发专家使用原子变量类,但 java.util.concurrent 类的很多可伸缩性改进都是由它们提供的。

    原子变量主要用于为原子地更新”热”字段提供有效的、细粒度的方式,”热”字段是指由多个线程频繁访问和更新的字段。另外,原子变量还是计数器或生成序号的自然机制。

  12. 性能与可伸缩性

    虽然 java.util.concurrent 努力的首要目标是使编写正确、线程安全的类更加容易,但它还有一个次要目标,就是提供可伸缩性。可伸缩性与性能完全不同,实际上,可伸缩性有时要以性能为代价来获得。

    性能是”可以快速执行此任务的程度”的评测。可伸缩性描述应用程序的吞吐量如何表现为它的工作量和可用计算资源增加。可伸缩的程序可以按比例使用更多的处理器、内存或 I/O 带宽来处理更多个工作量。当我们在并发环境中谈论可伸缩性时,我们是在问当许多线程同时访问给定类时,这个类的执行情况。

    java.util.concurrent 中的低级别类 ReentrantLock 和原子变量类的可伸缩性要比内置监视器(同步)锁定高得多。因此,使用 ReentrantLock 或原子变量类来协调共享访问的类也可能更具有可伸缩性。

  13. Hashtable 与 ConcurrentHashMap

    作为可伸缩性的例子,ConcurrentHashMap 实现设计的可伸缩性要比其线程安全的上一代 Hashtable 的可伸缩性强得多。Hashtable 一次只允许一个线程访问 Map;ConcurrentHashMap 允许多个读者并发执行,读者与写入者并发执行,以及一些写入者并发执行。因此,如果许多线程频繁访问共享映射,使用 ConcurrentHashMap 的总的吞吐量要比使用 Hashtable 的好。

    下表大致说明了 Hashtable 和 ConcurrentHashMap 之间的可伸缩性差别。在每次运行时,N 个线程并发执行紧密循环,它们从 Hashtable 或 ConcurrentHashMap 中检索随即关键字,60% 的失败检索将执行 put() 操作,2% 的成功检索执行 remove() 操作。测试在运行 Linux 的双处理器 Xeon 系统中执行。数据显示 10,000,000 个迭代的运行时间,对于 ConcurrentHashMap,标准化为一个线程的情况。可以看到直到许多线程,ConcurrentHashMap 的性能仍保持可伸缩性,而 Hashtable 的性能在出现锁定竞争时几乎立即下降。

    与通常的服务器应用程序相比,这个测试中的线程数看起来很少。然而,因为每个线程未进行其他操作,仅是重复地选择使用该表,所以这样可以模拟在执行一些实际工作的情况下使用该表的大量线程的竞争。

    线程 ConcurrentHashMap Hashtable
    1 1.0 1.51
    2 1.44 17.09
    4 1.83 29.9
    8 4.06 54.06
    16 7.5 119.44
    32 15.32 237.2
  14. Lock 与 synchronized 与原子

    下列基准说明了使用 java.util.concurrent 可能改进可伸缩性的例子。该基准将模拟旋转骰子,使用线性同余随机数生成器。有三个可用的随机数生成器的实现:一个使用同步来管理生成器的状态(单一变量),一个使用 ReentrantLock,另一个则使用 AtomicLong。下图显示了在 8-way Ultrasparc3 系统上,逐渐增加线程数量时这三个版本的相对吞吐量。(该图对原子变量方法的可伸缩性描述比较保守。)

  1. 公平与不公平

    java.util.concurrent 中许多类中的另外一个定制元素是”公平”的问题。公平锁定或公平信号是指在其中根据先进先出(FIFO)的原则给与线程锁定或信号。ReentrantLock、Semaphore 和 ReentrantReadWriteLock 的构造函数都可以使用变量确定锁定是否公平,或者是否允许闯入(线程获得锁定,即使它们等待的时间不是最长)。

    虽然闯入锁定的想法可能有些可笑,但实际上不公平、闯入的锁定非常普遍,且通常很受欢迎。使用同步访问的内置锁定不是公平锁定(且没有办法使它们公平)。相反,它们提供较弱的生病保证,要求所有线程最终都将获得锁定。

    大多数应用程序选择(且应该选择)闯入锁定而不是公平锁定的原因是性能。在大多数情况下,完全的公平不是程序正确性的要求,真正公平的成本相当高。下表向前面的面板中的表中添加了第四个数据集,并由一个公平锁定管理对 PRNG 状态的访问。注意闯入锁定与公平锁定之间吞吐量的巨大差别。

结束语

java.util.concurrent 包中包含大量有用的构建快,可以用它们来改进并发类的性能、可伸缩性、线程安全和可维护性。通过这些构建快,应该可以不再需要在您的代码中大量使用同步、wait/notify 和 Thread.start(),而用更高级别、标准化的、高性能并发实用程序来替换它们。

Exchanger,CyclicBarrier,Synchronizer

本文由 blog博主Caoer(草儿)原创,此处为转载。

由于原文两张图片不方便显示,这里暂时去掉,转载时本博(http://www.blogjava.net/mlh123caoer/archive/2007/10/24/155474.html)重新引用了图片,并调整了版面。

如原创作者认为本文侵权,请通知本博。

并发参考学习

ThreadPoolExecutor 源码分析

/*
* ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
*/

/*
*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
*/

package java.util.concurrent;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.*;

/**
*一个ExecutorService执行每个被提交入到线程池中的任务,通过Executors工厂方法进行配置。
*线程池处理两种不同的问题:通过减少每个任务调用的开销、提供边界和资源管理,包括线程,
*任务集合的执行,从而改进了执行大量异步任务时的性能问题。ThreadPoolExcecutor也维护着一
*些统计数据,如已完成任务的数目。
*面对一个提供了许多可调用参数和可扩展性的hooks.程序员通常比较喜欢用Executors的工厂方法。
*如Executors.newCachedThreadPool(无限大小的线程池,自动线程回收)、Executors.newFixedThreadPool
*(固定大小的线程池)、Executors.newSingleThreadExecutor(单个后台线程),为最常用的场
*景进行预配置。或者,使用这个类进行手动配置实现同样的效果的线程池,
*
* 核心和最大的线程池
*
* ThreadPoolExecutor会根据线程池的大小配置corePoolSize和maximumPoolSize来自动调整池的大小,
*可以通过getPoolSize查看池的大小。
*
* When a new task is submitted in method {@link #execute(Runnable)},
* and fewer than corePoolSize threads are running, a new thread is
* created to handle the request, even if other worker threads are
* idle. If there are more than corePoolSize but less than
* maximumPoolSize threads running, a new thread will be created only
* if the queue is full. By setting corePoolSize and maximumPoolSize
* the same, you create a fixed-size thread pool. By setting
* maximumPoolSize to an essentially unbounded value such as {@code
* Integer.MAX_VALUE}, you allow the pool to accommodate an arbitrary
* number of concurrent tasks. Most typically, core and maximum pool
* sizes are set only upon construction, but they may also be changed
* dynamically using {@link #setCorePoolSize} and {@link
* #setMaximumPoolSize}. </dd>
*
* <dt>On-demand construction</dt>
*
* <dd>By default, even core threads are initially created and
* started only when new tasks arrive, but this can be overridden
* dynamically using method {@link #prestartCoreThread} or {@link
* #prestartAllCoreThreads}. You probably want to prestart threads if
* you construct the pool with a non-empty queue. </dd>
*
* <dt>Creating new threads</dt>
*
* <dd>New threads are created using a {@link ThreadFactory}. If not
* otherwise specified, a {@link Executors#defaultThreadFactory} is
* used, that creates threads to all be in the same {@link
* ThreadGroup} and with the same {@code NORM_PRIORITY} priority and
* non-daemon status. By supplying a different ThreadFactory, you can
* alter the thread's name, thread group, priority, daemon status,
* etc. If a {@code ThreadFactory} fails to create a thread when asked
* by returning null from {@code newThread}, the executor will
* continue, but might not be able to execute any tasks. Threads
* should possess the "modifyThread" {@code RuntimePermission}. If
* worker threads or other threads using the pool do not possess this
* permission, service may be degraded: configuration changes may not
* take effect in a timely manner, and a shutdown pool may remain in a
* state in which termination is possible but not completed.</dd>
*
* <dt>Keep-alive times</dt>
*
* <dd>If the pool currently has more than corePoolSize threads,
* excess threads will be terminated if they have been idle for more
* than the keepAliveTime (see {@link #getKeepAliveTime(TimeUnit)}).
* This provides a means of reducing resource consumption when the
* pool is not being actively used. If the pool becomes more active
* later, new threads will be constructed. This parameter can also be
* changed dynamically using method {@link #setKeepAliveTime(long,
* TimeUnit)}. Using a value of {@code Long.MAX_VALUE} {@link
* TimeUnit#NANOSECONDS} effectively disables idle threads from ever
* terminating prior to shut down. By default, the keep-alive policy
* applies only when there are more than corePoolSize threads. But
* method {@link #allowCoreThreadTimeOut(boolean)} can be used to
* apply this time-out policy to core threads as well, so long as the
* keepAliveTime value is non-zero. </dd>
*
* <dt>Queuing</dt>
*
* <dd>Any {@link BlockingQueue} may be used to transfer and hold
* submitted tasks. The use of this queue interacts with pool sizing:
*
* <ul>
*
* <li> If fewer than corePoolSize threads are running, the Executor
* always prefers adding a new thread
* rather than queuing.</li>
*
* <li> If corePoolSize or more threads are running, the Executor
* always prefers queuing a request rather than adding a new
* thread.</li>
*
* <li> If a request cannot be queued, a new thread is created unless
* this would exceed maximumPoolSize, in which case, the task will be
* rejected.</li>
*
* </ul>
*
* There are three general strategies for queuing:
* <ol>
*
* <li> <em> Direct handoffs.</em> A good default choice for a work
* queue is a {@link SynchronousQueue} that hands off tasks to threads
* without otherwise holding them. Here, an attempt to queue a task
* will fail if no threads are immediately available to run it, so a
* new thread will be constructed. This policy avoids lockups when
* handling sets of requests that might have internal dependencies.
* Direct handoffs generally require unbounded maximumPoolSizes to
* avoid rejection of new submitted tasks. This in turn admits the
* possibility of unbounded thread growth when commands continue to
* arrive on average faster than they can be processed. </li>
*
* <li><em> Unbounded queues.</em> Using an unbounded queue (for
* example a {@link LinkedBlockingQueue} without a predefined
* capacity) will cause new tasks to wait in the queue when all
* corePoolSize threads are busy. Thus, no more than corePoolSize
* threads will ever be created. (And the value of the maximumPoolSize
* therefore doesn't have any effect.) This may be appropriate when
* each task is completely independent of others, so tasks cannot
* affect each others execution; for example, in a web page server.
* While this style of queuing can be useful in smoothing out
* transient bursts of requests, it admits the possibility of
* unbounded work queue growth when commands continue to arrive on
* average faster than they can be processed. </li>
*
* <li><em>Bounded queues.</em> A bounded queue (for example, an
* {@link ArrayBlockingQueue}) helps prevent resource exhaustion when
* used with finite maximumPoolSizes, but can be more difficult to
* tune and control. Queue sizes and maximum pool sizes may be traded
* off for each other: Using large queues and small pools minimizes
* CPU usage, OS resources, and context-switching overhead, but can
* lead to artificially low throughput. If tasks frequently block (for
* example if they are I/O bound), a system may be able to schedule
* time for more threads than you otherwise allow. Use of small queues
* generally requires larger pool sizes, which keeps CPUs busier but
* may encounter unacceptable scheduling overhead, which also
* decreases throughput. </li>
*
* </ol>
*
* </dd>
*
* <dt>Rejected tasks</dt>
*
* <dd>New tasks submitted in method {@link #execute(Runnable)} will be
* <em>rejected</em> when the Executor has been shut down, and also when
* the Executor uses finite bounds for both maximum threads and work queue
* capacity, and is saturated. In either case, the {@code execute} method
* invokes the {@link
* RejectedExecutionHandler#rejectedExecution(Runnable, ThreadPoolExecutor)}
* method of its {@link RejectedExecutionHandler}. Four predefined handler
* policies are provided:
*
* <ol>
*
* <li> In the default {@link ThreadPoolExecutor.AbortPolicy}, the
* handler throws a runtime {@link RejectedExecutionException} upon
* rejection. </li>
*
* <li> In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread
* that invokes {@code execute} itself runs the task. This provides a
* simple feedback control mechanism that will slow down the rate that
* new tasks are submitted. </li>
*
* <li> In {@link ThreadPoolExecutor.DiscardPolicy}, a task that
* cannot be executed is simply dropped. </li>
*
* <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the
* executor is not shut down, the task at the head of the work queue
* is dropped, and then execution is retried (which can fail again,
* causing this to be repeated.) </li>
*
* </ol>
*
* It is possible to define and use other kinds of {@link
* RejectedExecutionHandler} classes. Doing so requires some care
* especially when policies are designed to work only under particular
* capacity or queuing policies. </dd>
*
* <dt>Hook methods</dt>
*
* <dd>This class provides {@code protected} overridable
* {@link #beforeExecute(Thread, Runnable)} and
* {@link #afterExecute(Runnable, Throwable)} methods that are called
* before and after execution of each task. These can be used to
* manipulate the execution environment; for example, reinitializing
* ThreadLocals, gathering statistics, or adding log entries.
* Additionally, method {@link #terminated} can be overridden to perform
* any special processing that needs to be done once the Executor has
* fully terminated.
*
* <p>If hook or callback methods throw exceptions, internal worker
* threads may in turn fail and abruptly terminate.</dd>
*
* <dt>Queue maintenance</dt>
*
* <dd>Method {@link #getQueue()} allows access to the work queue
* for purposes of monitoring and debugging. Use of this method for
* any other purpose is strongly discouraged. Two supplied methods,
* {@link #remove(Runnable)} and {@link #purge} are available to
* assist in storage reclamation when large numbers of queued tasks
* become cancelled.</dd>
*
* <dt>Finalization</dt>
*
* <dd>A pool that is no longer referenced in a program <em>AND</em>
* has no remaining threads will be {@code shutdown} automatically. If
* you would like to ensure that unreferenced pools are reclaimed even
* if users forget to call {@link #shutdown}, then you must arrange
* that unused threads eventually die, by setting appropriate
* keep-alive times, using a lower bound of zero core threads and/or
* setting {@link #allowCoreThreadTimeOut(boolean)}. </dd>
*
* </dl>
*
* <p><b>Extension example</b>. Most extensions of this class
* override one or more of the protected hook methods. For example,
* here is a subclass that adds a simple pause/resume feature:
*
* <pre> {@code
* class PausableThreadPoolExecutor extends ThreadPoolExecutor {
* private boolean isPaused;
* private ReentrantLock pauseLock = new ReentrantLock();
* private Condition unpaused = pauseLock.newCondition();
*
* public PausableThreadPoolExecutor(...) { super(...); }
*
* protected void beforeExecute(Thread t, Runnable r) {
* super.beforeExecute(t, r);
* pauseLock.lock();
* try {
* while (isPaused) unpaused.await();
* } catch (InterruptedException ie) {
* t.interrupt();
* } finally {
* pauseLock.unlock();
* }
* }
*
* public void pause() {
* pauseLock.lock();
* try {
* isPaused = true;
* } finally {
* pauseLock.unlock();
* }
* }
*
* public void resume() {
* pauseLock.lock();
* try {
* isPaused = false;
* unpaused.signalAll();
* } finally {
* pauseLock.unlock();
* }
* }
* }}</pre>
*
* @since 1.5
* @author Doug Lea
*/
public class ThreadPoolExecutor extends AbstractExecutorService {
/**
* The main pool control state, ctl, is an atomic integer packing
* two conceptual fields
* workerCount, indicating the effective number of threads
* runState, indicating whether running, shutting down etc
*
* In order to pack them into one int, we limit workerCount to
* (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
* billion) otherwise representable. If this is ever an issue in
* the future, the variable can be changed to be an AtomicLong,
* and the shift/mask constants below adjusted. But until the need
* arises, this code is a bit faster and simpler using an int.
*
* The workerCount is the number of workers that have been
* permitted to start and not permitted to stop. The value may be
* transiently different from the actual number of live threads,
* for example when a ThreadFactory fails to create a thread when
* asked, and when exiting threads are still performing
* bookkeeping before terminating. The user-visible pool size is
* reported as the current size of the workers set.
*
* The runState provides the main lifecycle control, taking on values:
*
* RUNNING: Accept new tasks and process queued tasks
* SHUTDOWN: Don't accept new tasks, but process queued tasks
* STOP: Don't accept new tasks, don't process queued tasks,
* and interrupt in-progress tasks
* TIDYING: All tasks have terminated, workerCount is zero,
* the thread transitioning to state TIDYING
* will run the terminated() hook method
* TERMINATED: terminated() has completed
*
* The numerical order among these values matters, to allow
* ordered comparisons. The runState monotonically increases over
* time, but need not hit each state. The transitions are:
*
* RUNNING -> SHUTDOWN
* On invocation of shutdown(), perhaps implicitly in finalize()
* (RUNNING or SHUTDOWN) -> STOP
* On invocation of shutdownNow()
* SHUTDOWN -> TIDYING
* When both queue and pool are empty
* STOP -> TIDYING
* When pool is empty
* TIDYING -> TERMINATED
* When the terminated() hook method has completed
*
* Threads waiting in awaitTermination() will return when the
* state reaches TERMINATED.
*
* Detecting the transition from SHUTDOWN to TIDYING is less
* straightforward than you'd like because the queue may become
* empty after non-empty and vice versa during SHUTDOWN state, but
* we can only terminate if, after seeing that it is empty, we see
* that workerCount is 0 (which sometimes entails a recheck -- see
* below).
*/
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; }

/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/

private static boolean runStateLessThan(int c, int s) {
return c < s;
}

private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}

private static boolean isRunning(int c) {
return c < SHUTDOWN;
}

/**
* Attempts to CAS-increment the workerCount field of ctl.
*/
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}

/**
* Attempts to CAS-decrement the workerCount field of ctl.
*/
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}

/**
* Decrements the workerCount field of ctl. This is called only on
* abrupt termination of a thread (see processWorkerExit). Other
* decrements are performed within getTask.
*/
private void decrementWorkerCount() {
do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}

/**
* The queue used for holding tasks and handing off to worker
* threads. We do not require that workQueue.poll() returning
* null necessarily means that workQueue.isEmpty(), so rely
* solely on isEmpty to see if the queue is empty (which we must
* do for example when deciding whether to transition from
* SHUTDOWN to TIDYING). This accommodates special-purpose
* queues such as DelayQueues for which poll() is allowed to
* return null even if it may later return non-null when delays
* expire.
*/
private final BlockingQueue<Runnable> workQueue;

/**
* Lock held on access to workers set and related bookkeeping.
* While we could use a concurrent set of some sort, it turns out
* to be generally preferable to use a lock. Among the reasons is
* that this serializes interruptIdleWorkers, which avoids
* unnecessary interrupt storms, especially during shutdown.
* Otherwise exiting threads would concurrently interrupt those
* that have not yet interrupted. It also simplifies some of the
* associated statistics bookkeeping of largestPoolSize etc. We
* also hold mainLock on shutdown and shutdownNow, for the sake of
* ensuring workers set is stable while separately checking
* permission to interrupt and actually interrupting.
*/
private final ReentrantLock mainLock = new ReentrantLock();

/**
* Set containing all worker threads in pool. Accessed only when
* holding mainLock.
*/
private final HashSet<Worker> workers = new HashSet<Worker>();

/**
* Wait condition to support awaitTermination
*/
private final Condition termination = mainLock.newCondition();

/**
* Tracks largest attained pool size. Accessed only under
* mainLock.
*/
private int largestPoolSize;

/**
* Counter for completed tasks. Updated only on termination of
* worker threads. Accessed only under mainLock.
*/
private long completedTaskCount;

/*
* All user control parameters are declared as volatiles so that
* ongoing actions are based on freshest values, but without need
* for locking, since no internal invariants depend on them
* changing synchronously with respect to other actions.
*/

/**
* Factory for new threads. All threads are created using this
* factory (via method addWorker). All callers must be prepared
* for addWorker to fail, which may reflect a system or user's
* policy limiting the number of threads. Even though it is not
* treated as an error, failure to create threads may result in
* new tasks being rejected or existing ones remaining stuck in
* the queue.
*
* We go further and preserve pool invariants even in the face of
* errors such as OutOfMemoryError, that might be thrown while
* trying to create threads. Such errors are rather common due to
* the need to allocate a native stack in Thread.start, and users
* will want to perform clean pool shutdown to clean up. There
* will likely be enough memory available for the cleanup code to
* complete without encountering yet another OutOfMemoryError.
*/
private volatile ThreadFactory threadFactory;

/**
* Handler called when saturated or shutdown in execute.
*/
private volatile RejectedExecutionHandler handler;

/**
* Timeout in nanoseconds for idle threads waiting for work.
* Threads use this timeout when there are more than corePoolSize
* present or if allowCoreThreadTimeOut. Otherwise they wait
* forever for new work.
*/
private volatile long keepAliveTime;

/**
* If false (default), core threads stay alive even when idle.
* If true, core threads use keepAliveTime to time out waiting
* for work.
*/
private volatile boolean allowCoreThreadTimeOut;

/**
* Core pool size is the minimum number of workers to keep alive
* (and not allow to time out etc) unless allowCoreThreadTimeOut
* is set, in which case the minimum is zero.
*/
private volatile int corePoolSize;

/**
* Maximum pool size. Note that the actual maximum is internally
* bounded by CAPACITY.
*/
private volatile int maximumPoolSize;

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

/**
* Permission required for callers of shutdown and shutdownNow.
* We additionally require (see checkShutdownAccess) that callers
* have permission to actually interrupt threads in the worker set
* (as governed by Thread.interrupt, which relies on
* ThreadGroup.checkAccess, which in turn relies on
* SecurityManager.checkAccess). Shutdowns are attempted only if
* these checks pass.
*
* All actual invocations of Thread.interrupt (see
* interruptIdleWorkers and interruptWorkers) ignore
* SecurityExceptions, meaning that the attempted interrupts
* silently fail. In the case of shutdown, they should not fail
* unless the SecurityManager has inconsistent policies, sometimes
* allowing access to a thread and sometimes not. In such cases,
* failure to actually interrupt threads may disable or delay full
* termination. Other uses of interruptIdleWorkers are advisory,
* and failure to actually interrupt will merely delay response to
* configuration changes so is not handled exceptionally.
*/
private static final RuntimePermission shutdownPerm =
new RuntimePermission("modifyThread");

/**
* Class Worker mainly maintains interrupt control state for
* threads running tasks, along with other minor bookkeeping.
* This class opportunistically extends AbstractQueuedSynchronizer
* to simplify acquiring and releasing a lock surrounding each
* task execution. This protects against interrupts that are
* intended to wake up a worker thread waiting for a task from
* instead interrupting a task being run. We implement a simple
* non-reentrant mutual exclusion lock rather than use
* ReentrantLock because we do not want worker tasks to be able to
* reacquire the lock when they invoke pool control methods like
* setCorePoolSize. Additionally, to suppress interrupts until
* the thread actually starts running tasks, we initialize lock
* state to a negative value, and clear it upon start (in
* runWorker).
*/
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;

/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;

/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
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);
}

// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.

protected boolean isHeldExclusively() {
return getState() != 0;
}

protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}

protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}

public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }

void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}

/*
* Methods for setting control state
*/

/**
* Transitions runState to given target, or leaves it alone if
* already at least the given target.
*
* @param targetState the desired state, either SHUTDOWN or STOP
* (but not TIDYING or TERMINATED -- use tryTerminate for that)
*/
private void advanceRunState(int targetState) {
for (;;) {
int c = ctl.get();
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}

/**
* Transitions to TERMINATED state if either (SHUTDOWN and pool
* and queue empty) or (STOP and pool empty). If otherwise
* eligible to terminate but workerCount is nonzero, interrupts an
* idle worker to ensure that shutdown signals propagate. This
* method must be called following any action that might make
* termination possible -- reducing worker count or removing tasks
* from the queue during shutdown. The method is non-private to
* allow access from ScheduledThreadPoolExecutor.
*/
final void tryTerminate() {
for (;;) {
int c = ctl.get();
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}

/*
* Methods for controlling interrupts to worker threads.
*/

/**
* If there is a security manager, makes sure caller has
* permission to shut down threads in general (see shutdownPerm).
* If this passes, additionally makes sure the caller is allowed
* to interrupt each worker thread. This might not be true even if
* first check passed, if the SecurityManager treats some threads
* specially.
*/
private void checkShutdownAccess() {
SecurityManager security = System.getSecurityManager();
if (security != null) {
security.checkPermission(shutdownPerm);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
security.checkAccess(w.thread);
} finally {
mainLock.unlock();
}
}
}

/**
* Interrupts all threads, even if active. Ignores SecurityExceptions
* (in which case some threads may remain uninterrupted).
*/
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}

/**
* Interrupts threads that might be waiting for tasks (as
* indicated by not being locked) so they can check for
* termination or configuration changes. Ignores
* SecurityExceptions (in which case some threads may remain
* uninterrupted).
*
* @param onlyOne If true, interrupt at most one worker. This is
* called only from tryTerminate when termination is otherwise
* enabled but there are still other workers. In this case, at
* most one waiting worker is interrupted to propagate shutdown
* signals in case all threads are currently waiting.
* Interrupting any arbitrary thread ensures that newly arriving
* workers since shutdown began will also eventually exit.
* To guarantee eventual termination, it suffices to always
* interrupt only one idle worker, but shutdown() interrupts all
* idle workers so that redundant workers exit promptly, not
* waiting for a straggler task to finish.
*/
private void interruptIdleWorkers(boolean onlyOne) {
final 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();
}
}

/**
* Common form of interruptIdleWorkers, to avoid having to
* remember what the boolean argument means.
*/
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}

private static final boolean ONLY_ONE = true;

/*
* Misc utilities, most of which are also exported to
* ScheduledThreadPoolExecutor
*/

/**
* Invokes the rejected execution handler for the given command.
* Package-protected for use by ScheduledThreadPoolExecutor.
*/
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}

/**
* Performs any further cleanup following run state transition on
* invocation of shutdown. A no-op here, but used by
* ScheduledThreadPoolExecutor to cancel delayed tasks.
*/
void onShutdown() {
}

/**
* State check needed by ScheduledThreadPoolExecutor to
* enable running tasks during shutdown.
*
* @param shutdownOK true if should return true if SHUTDOWN
*/
final boolean isRunningOrShutdown(boolean shutdownOK) {
int rs = runStateOf(ctl.get());
return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
}

/**
* Drains the task queue into a new list, normally using
* drainTo. But if the queue is a DelayQueue or any other kind of
* queue for which poll or drainTo may fail to remove some
* elements, it deletes them one by one.
*/
private List<Runnable> drainQueue() {
BlockingQueue<Runnable> q = workQueue;
ArrayList<Runnable> taskList = new ArrayList<Runnable>();
q.drainTo(taskList);
if (!q.isEmpty()) {
for (Runnable r : q.toArray(new Runnable[0])) {
if (q.remove(r))
taskList.add(r);
}
}
return taskList;
}

/*
* Methods for creating, running and cleaning up after workers
*/

/**
* Checks if a new worker can be added with respect to current
* pool state and the given bound (either core or maximum). If so,
* the worker count is adjusted accordingly, and, if possible, a
* new worker is created and started, running firstTask as its
* first task. This method returns false if the pool is stopped or
* eligible to shut down. It also returns false if the thread
* factory fails to create a thread when asked. If the thread
* creation fails, either due to the thread factory returning
* null, or due to an exception (typically OutOfMemoryError in
* Thread.start()), we roll back cleanly.
*
* @param firstTask the task the new thread should run first (or
* null if none). Workers are created with an initial first task
* (in method execute()) to bypass queuing when there are fewer
* than corePoolSize threads (in which case we always start one),
* or when the queue is full (in which case we must bypass queue).
* Initially idle threads are usually created via
* prestartCoreThread or to replace other dying workers.
*
* @param core if true use corePoolSize as bound, else
* maximumPoolSize. (A boolean indicator is used here rather than a
* value to ensure reads of fresh values after checking other pool
* state).
* @return true if successful
*/
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;
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();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

/**
* Rolls back the worker thread creation.
* - removes worker from workers, if present
* - decrements worker count
* - rechecks for termination, in case the existence of this
* worker was holding up termination
*/
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}

/**
* Performs cleanup and bookkeeping for a dying worker. Called
* only from worker threads. Unless completedAbruptly is set,
* assumes that workerCount has already been adjusted to account
* for exit. This method removes thread from worker set, and
* possibly terminates the pool or replaces the worker if either
* it exited due to user task exception or if fewer than
* corePoolSize workers are running or queue is non-empty but
* there are no workers.
*
* @param w the worker
* @param completedAbruptly if the worker died due to user exception
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}

tryTerminate();

int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}

/**
* Performs blocking or timed wait for a task, depending on
* current configuration settings, or returns null if this worker
* must exit because of any of:
* 1. There are more than maximumPoolSize workers (due to
* a call to setMaximumPoolSize).
* 2. The pool is stopped.
* 3. The pool is shutdown and the queue is empty.
* 4. This worker timed out waiting for a task, and timed-out
* workers are subject to termination (that is,
* {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
* both before and after the timed wait, and if the queue is
* non-empty, this worker is not the last thread in the pool.
*
* @return task, or null if the worker must exit, in which case
* workerCount is decremented
*/
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;
}
}
}

/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
*
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is
* running, we get tasks from getTask. If it returns null then the
* worker exits due to changed pool state or configuration
* parameters. Other exits result from exception throws in
* external code, in which case completedAbruptly holds, which
* usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent
* other pool interrupts while the task is executing, and then we
* ensure that unless pool is stopping, this thread does not have
* its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which
* might throw an exception, in which case we cause thread to die
* (breaking loop with completedAbruptly true) without processing
* the task.
*
* 4. Assuming beforeExecute completes normally, we run the task,
* gathering any of its thrown exceptions to send to afterExecute.
* We separately handle RuntimeException, Error (both of which the
* specs guarantee that we trap) and arbitrary Throwables.
* Because we cannot rethrow Throwables within Runnable.run, we
* wrap them within Errors on the way out (to the thread's
* UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate
* information as we can provide about any problems encountered by
* user code.
*
* @param w the worker
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
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);
}
}

// Public constructors and methods

/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default thread factory and rejected execution handler.
* It may be more convenient to use one of the {@link Executors} factory
* methods instead of this general purpose constructor.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}

/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default rejected execution handler.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}

/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default thread factory.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code handler} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}

/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} or {@code handler} is null
*/
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;
}

/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
*
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
*
* @param command the task to execute
* @throws RejectedExecutionException at discretion of
* {@code RejectedExecutionHandler}, if the task
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/
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) {
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);
}

/**
* Initiates an orderly shutdown in which previously submitted
* tasks are executed, but no new tasks will be accepted.
* Invocation has no additional effect if already shut down.
*
* <p>This method does not wait for previously submitted tasks to
* complete execution. Use {@link #awaitTermination awaitTermination}
* to do that.
*
* @throws SecurityException {@inheritDoc}
*/
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}

/**
* Attempts to stop all actively executing tasks, halts the
* processing of waiting tasks, and returns a list of the tasks
* that were awaiting execution. These tasks are drained (removed)
* from the task queue upon return from this method.
*
* <p>This method does not wait for actively executing tasks to
* terminate. Use {@link #awaitTermination awaitTermination} to
* do that.
*
* <p>There are no guarantees beyond best-effort attempts to stop
* processing actively executing tasks. This implementation
* cancels tasks via {@link Thread#interrupt}, so any task that
* fails to respond to interrupts may never terminate.
*
* @throws SecurityException {@inheritDoc}
*/
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}

public boolean isShutdown() {
return ! isRunning(ctl.get());
}

/**
* Returns true if this executor is in the process of terminating
* after {@link #shutdown} or {@link #shutdownNow} but has not
* completely terminated. This method may be useful for
* debugging. A return of {@code true} reported a sufficient
* period after shutdown may indicate that submitted tasks have
* ignored or suppressed interruption, causing this executor not
* to properly terminate.
*
* @return {@code true} if terminating but not yet terminated
*/
public boolean isTerminating() {
int c = ctl.get();
return ! isRunning(c) && runStateLessThan(c, TERMINATED);
}

public boolean isTerminated() {
return runStateAtLeast(ctl.get(), TERMINATED);
}

public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (;;) {
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}

/**
* Invokes {@code shutdown} when this executor is no longer
* referenced and it has no threads.
*/
protected void finalize() {
shutdown();
}

/**
* Sets the thread factory used to create new threads.
*
* @param threadFactory the new thread factory
* @throws NullPointerException if threadFactory is null
* @see #getThreadFactory
*/
public void setThreadFactory(ThreadFactory threadFactory) {
if (threadFactory == null)
throw new NullPointerException();
this.threadFactory = threadFactory;
}

/**
* Returns the thread factory used to create new threads.
*
* @return the current thread factory
* @see #setThreadFactory(ThreadFactory)
*/
public ThreadFactory getThreadFactory() {
return threadFactory;
}

/**
* Sets a new handler for unexecutable tasks.
*
* @param handler the new handler
* @throws NullPointerException if handler is null
* @see #getRejectedExecutionHandler
*/
public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
if (handler == null)
throw new NullPointerException();
this.handler = handler;
}

/**
* Returns the current handler for unexecutable tasks.
*
* @return the current handler
* @see #setRejectedExecutionHandler(RejectedExecutionHandler)
*/
public RejectedExecutionHandler getRejectedExecutionHandler() {
return handler;
}

/**
* Sets the core number of threads. This overrides any value set
* in the constructor. If the new value is smaller than the
* current value, excess existing threads will be terminated when
* they next become idle. If larger, new threads will, if needed,
* be started to execute any queued tasks.
*
* @param corePoolSize the new core size
* @throws IllegalArgumentException if {@code corePoolSize < 0}
* @see #getCorePoolSize
*/
public void setCorePoolSize(int corePoolSize) {
if (corePoolSize < 0)
throw new IllegalArgumentException();
int delta = corePoolSize - this.corePoolSize;
this.corePoolSize = corePoolSize;
if (workerCountOf(ctl.get()) > corePoolSize)
interruptIdleWorkers();
else if (delta > 0) {
// We don't really know how many new threads are "needed".
// As a heuristic, prestart enough new workers (up to new
// core size) to handle the current number of tasks in
// queue, but stop if queue becomes empty while doing so.
int k = Math.min(delta, workQueue.size());
while (k-- > 0 && addWorker(null, true)) {
if (workQueue.isEmpty())
break;
}
}
}

/**
* Returns the core number of threads.
*
* @return the core number of threads
* @see #setCorePoolSize
*/
public int getCorePoolSize() {
return corePoolSize;
}

/**
* Starts a core thread, causing it to idly wait for work. This
* overrides the default policy of starting core threads only when
* new tasks are executed. This method will return {@code false}
* if all core threads have already been started.
*
* @return {@code true} if a thread was started
*/
public boolean prestartCoreThread() {
return workerCountOf(ctl.get()) < corePoolSize &&
addWorker(null, true);
}

/**
* Same as prestartCoreThread except arranges that at least one
* thread is started even if corePoolSize is 0.
*/
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(null, false);
}

/**
* Starts all core threads, causing them to idly wait for work. This
* overrides the default policy of starting core threads only when
* new tasks are executed.
*
* @return the number of threads started
*/
public int prestartAllCoreThreads() {
int n = 0;
while (addWorker(null, true))
++n;
return n;
}

/**
* Returns true if this pool allows core threads to time out and
* terminate if no tasks arrive within the keepAlive time, being
* replaced if needed when new tasks arrive. When true, the same
* keep-alive policy applying to non-core threads applies also to
* core threads. When false (the default), core threads are never
* terminated due to lack of incoming tasks.
*
* @return {@code true} if core threads are allowed to time out,
* else {@code false}
*
* @since 1.6
*/
public boolean allowsCoreThreadTimeOut() {
return allowCoreThreadTimeOut;
}

/**
* Sets the policy governing whether core threads may time out and
* terminate if no tasks arrive within the keep-alive time, being
* replaced if needed when new tasks arrive. When false, core
* threads are never terminated due to lack of incoming
* tasks. When true, the same keep-alive policy applying to
* non-core threads applies also to core threads. To avoid
* continual thread replacement, the keep-alive time must be
* greater than zero when setting {@code true}. This method
* should in general be called before the pool is actively used.
*
* @param value {@code true} if should time out, else {@code false}
* @throws IllegalArgumentException if value is {@code true}
* and the current keep-alive time is not greater than zero
*
* @since 1.6
*/
public void allowCoreThreadTimeOut(boolean value) {
if (value && keepAliveTime <= 0)
throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
if (value != allowCoreThreadTimeOut) {
allowCoreThreadTimeOut = value;
if (value)
interruptIdleWorkers();
}
}

/**
* Sets the maximum allowed number of threads. This overrides any
* value set in the constructor. If the new value is smaller than
* the current value, excess existing threads will be
* terminated when they next become idle.
*
* @param maximumPoolSize the new maximum
* @throws IllegalArgumentException if the new maximum is
* less than or equal to zero, or
* less than the {@linkplain #getCorePoolSize core pool size}
* @see #getMaximumPoolSize
*/
public void setMaximumPoolSize(int maximumPoolSize) {
if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
throw new IllegalArgumentException();
this.maximumPoolSize = maximumPoolSize;
if (workerCountOf(ctl.get()) > maximumPoolSize)
interruptIdleWorkers();
}

/**
* Returns the maximum allowed number of threads.
*
* @return the maximum allowed number of threads
* @see #setMaximumPoolSize
*/
public int getMaximumPoolSize() {
return maximumPoolSize;
}

/**
* Sets the time limit for which threads may remain idle before
* being terminated. If there are more than the core number of
* threads currently in the pool, after waiting this amount of
* time without processing a task, excess threads will be
* terminated. This overrides any value set in the constructor.
*
* @param time the time to wait. A time value of zero will cause
* excess threads to terminate immediately after executing tasks.
* @param unit the time unit of the {@code time} argument
* @throws IllegalArgumentException if {@code time} less than zero or
* if {@code time} is zero and {@code allowsCoreThreadTimeOut}
* @see #getKeepAliveTime(TimeUnit)
*/
public void setKeepAliveTime(long time, TimeUnit unit) {
if (time < 0)
throw new IllegalArgumentException();
if (time == 0 && allowsCoreThreadTimeOut())
throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
long keepAliveTime = unit.toNanos(time);
long delta = keepAliveTime - this.keepAliveTime;
this.keepAliveTime = keepAliveTime;
if (delta < 0)
interruptIdleWorkers();
}

/**
* Returns the thread keep-alive time, which is the amount of time
* that threads in excess of the core pool size may remain
* idle before being terminated.
*
* @param unit the desired time unit of the result
* @return the time limit
* @see #setKeepAliveTime(long, TimeUnit)
*/
public long getKeepAliveTime(TimeUnit unit) {
return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
}

/* User-level queue utilities */

/**
* Returns the task queue used by this executor. Access to the
* task queue is intended primarily for debugging and monitoring.
* This queue may be in active use. Retrieving the task queue
* does not prevent queued tasks from executing.
*
* @return the task queue
*/
public BlockingQueue<Runnable> getQueue() {
return workQueue;
}

/**
* Removes this task from the executor's internal queue if it is
* present, thus causing it not to be run if it has not already
* started.
*
* <p>This method may be useful as one part of a cancellation
* scheme. It may fail to remove tasks that have been converted
* into other forms before being placed on the internal queue. For
* example, a task entered using {@code submit} might be
* converted into a form that maintains {@code Future} status.
* However, in such cases, method {@link #purge} may be used to
* remove those Futures that have been cancelled.
*
* @param task the task to remove
* @return {@code true} if the task was removed
*/
public boolean remove(Runnable task) {
boolean removed = workQueue.remove(task);
tryTerminate(); // In case SHUTDOWN and now empty
return removed;
}

/**
* Tries to remove from the work queue all {@link Future}
* tasks that have been cancelled. This method can be useful as a
* storage reclamation operation, that has no other impact on
* functionality. Cancelled tasks are never executed, but may
* accumulate in work queues until worker threads can actively
* remove them. Invoking this method instead tries to remove them now.
* However, this method may fail to remove tasks in
* the presence of interference by other threads.
*/
public void purge() {
final BlockingQueue<Runnable> q = workQueue;
try {
Iterator<Runnable> it = q.iterator();
while (it.hasNext()) {
Runnable r = it.next();
if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
it.remove();
}
} catch (ConcurrentModificationException fallThrough) {
// Take slow path if we encounter interference during traversal.
// Make copy for traversal and call remove for cancelled entries.
// The slow path is more likely to be O(N*N).
for (Object r : q.toArray())
if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
q.remove(r);
}

tryTerminate(); // In case SHUTDOWN and now empty
}

/* Statistics */

/**
* Returns the current number of threads in the pool.
*
* @return the number of threads
*/
public int getPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Remove rare and surprising possibility of
// isTerminated() && getPoolSize() > 0
return runStateAtLeast(ctl.get(), TIDYING) ? 0
: workers.size();
} finally {
mainLock.unlock();
}
}

/**
* Returns the approximate number of threads that are actively
* executing tasks.
*
* @return the number of threads
*/
public int getActiveCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int n = 0;
for (Worker w : workers)
if (w.isLocked())
++n;
return n;
} finally {
mainLock.unlock();
}
}

/**
* Returns the largest number of threads that have ever
* simultaneously been in the pool.
*
* @return the number of threads
*/
public int getLargestPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
return largestPoolSize;
} finally {
mainLock.unlock();
}
}

/**
* Returns the approximate total number of tasks that have ever been
* scheduled for execution. Because the states of tasks and
* threads may change dynamically during computation, the returned
* value is only an approximation.
*
* @return the number of tasks
*/
public long getTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers) {
n += w.completedTasks;
if (w.isLocked())
++n;
}
return n + workQueue.size();
} finally {
mainLock.unlock();
}
}

/**
* Returns the approximate total number of tasks that have
* completed execution. Because the states of tasks and threads
* may change dynamically during computation, the returned value
* is only an approximation, but one that does not ever decrease
* across successive calls.
*
* @return the number of tasks
*/
public long getCompletedTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers)
n += w.completedTasks;
return n;
} finally {
mainLock.unlock();
}
}

/**
* Returns a string identifying this pool, as well as its state,
* including indications of run state and estimated worker and
* task counts.
*
* @return a string identifying this pool, as well as its state
*/
public String toString() {
long ncompleted;
int nworkers, nactive;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
ncompleted = completedTaskCount;
nactive = 0;
nworkers = workers.size();
for (Worker w : workers) {
ncompleted += w.completedTasks;
if (w.isLocked())
++nactive;
}
} finally {
mainLock.unlock();
}
int c = ctl.get();
String rs = (runStateLessThan(c, SHUTDOWN) ? "Running" :
(runStateAtLeast(c, TERMINATED) ? "Terminated" :
"Shutting down"));
return super.toString() +
"[" + rs +
", pool size = " + nworkers +
", active threads = " + nactive +
", queued tasks = " + workQueue.size() +
", completed tasks = " + ncompleted +
"]";
}

/* Extension hooks */

/**
* Method invoked prior to executing the given Runnable in the
* given thread. This method is invoked by thread {@code t} that
* will execute task {@code r}, and may be used to re-initialize
* ThreadLocals, or to perform logging.
*
* <p>This implementation does nothing, but may be customized in
* subclasses. Note: To properly nest multiple overridings, subclasses
* should generally invoke {@code super.beforeExecute} at the end of
* this method.
*
* @param t the thread that will run task {@code r}
* @param r the task that will be executed
*/
protected void beforeExecute(Thread t, Runnable r) { }

/**
* Method invoked upon completion of execution of the given Runnable.
* This method is invoked by the thread that executed the task. If
* non-null, the Throwable is the uncaught {@code RuntimeException}
* or {@code Error} that caused execution to terminate abruptly.
*
* <p>This implementation does nothing, but may be customized in
* subclasses. Note: To properly nest multiple overridings, subclasses
* should generally invoke {@code super.afterExecute} at the
* beginning of this method.
*
* <p><b>Note:</b> When actions are enclosed in tasks (such as
* {@link FutureTask}) either explicitly or via methods such as
* {@code submit}, these task objects catch and maintain
* computational exceptions, and so they do not cause abrupt
* termination, and the internal exceptions are <em>not</em>
* passed to this method. If you would like to trap both kinds of
* failures in this method, you can further probe for such cases,
* as in this sample subclass that prints either the direct cause
* or the underlying exception if a task has been aborted:
*
* <pre> {@code
* class ExtendedExecutor extends ThreadPoolExecutor {
* // ...
* protected void afterExecute(Runnable r, Throwable t) {
* super.afterExecute(r, t);
* if (t == null && r instanceof Future<?>) {
* try {
* Object result = ((Future<?>) r).get();
* } catch (CancellationException ce) {
* t = ce;
* } catch (ExecutionException ee) {
* t = ee.getCause();
* } catch (InterruptedException ie) {
* Thread.currentThread().interrupt(); // ignore/reset
* }
* }
* if (t != null)
* System.out.println(t);
* }
* }}</pre>
*
* @param r the runnable that has completed
* @param t the exception that caused termination, or null if
* execution completed normally
*/
protected void afterExecute(Runnable r, Throwable t) { }

/**
* Method invoked when the Executor has terminated. Default
* implementation does nothing. Note: To properly nest multiple
* overridings, subclasses should generally invoke
* {@code super.terminated} within this method.
*/
protected void terminated() { }

/* Predefined RejectedExecutionHandlers */

/**
* A handler for rejected tasks that runs the rejected task
* directly in the calling thread of the {@code execute} method,
* unless the executor has been shut down, in which case the task
* is discarded.
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/
public CallerRunsPolicy() { }

/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}

/**
* A handler for rejected tasks that throws a
* {@code RejectedExecutionException}.
*/
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() { }

/**
* Always throws RejectedExecutionException.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
* @throws RejectedExecutionException always
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}

/**
* A handler for rejected tasks that silently discards the
* rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }

/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}

/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
* is shut down, in which case the task is discarded.
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() { }

/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
}