Java concurrent Framework并发之Executor框架分析(核心的地方要多次研究)

现在的位置: 首页 > 综合 > 正文

Java concurrent Framework并发之Executor框架分析(核心的地方要多次研究)

2019年09月28日 ⁄ 综合 ⁄ 共 35076字 ⁄ 字号小中大 ⁄ 评论关闭

文章目录

ThreadPoolExecutor简介

public class ThreadPoolExecutor extends AbstractExecutorService
一个ExecutorService，它使用可能的几个池线程之一执行每个提交的任务，通常使用Executors工厂方法配置。

线程池可以解决两个不同问题：由于减少了每个任务调用的开销，它们通常可以在执行大量异步任务时提供增强的性能，并且还可以提供绑定和管理资源（包括执行任务集时使用的线程）的方法。每个ThreadPoolExecutor还维护着一些基本的统计数据，如完成的任务数。

Executors工厂方法

     为了便于跨大量上下文使用，此类提供了很多可调整的参数和扩展钩子 (hook)。但是，强烈建议程序员使用较为方便的 Executors工厂方法:
     Executors.newCachedThreadPool()（无界线程池，可以进行自动线程回收）、
     Executors.newFixedThreadPool(int)（固定大小线程池）、
     Executors.newSingleThreadExecutor()（单个后台线程），它们均为大多数使用场景预定义了设置。否则，在手动配置和调整此类时，使用以下指导：

ThreadPoolExecutor构造函数

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

核心大小corePoolSize和最大池大小maximumPoolSize

ThreadPoolExecutor将根据corePoolSize（参见 getCorePoolSize()）和maximumPoolSize（参见 getMaximumPoolSize()）设置的边界自动调整池大小。当新任务在方法execute(java.lang.Runnable)中提交时，如果运行的线程少于corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的。如果运行的线程多于corePoolSize而少于maximumPoolSize，则仅当队列满时才创建新线程。如果设置的corePoolSize和maximumPoolSize相同，则创建了固定大小的线程池。如果将maximumPoolSize设置为基本的无界值（如
Integer.MAX_VALUE），则允许池适应任意数量的并发任务。在大多数情况下，核心大小corePoolSize和最大池大小maximumPoolSize仅基于构造来设置，不过也可以使用setCorePoolSize(int) 和setMaximumPoolSize(int)进行动态更改。

按需构造 On-demand construction

默认情况下，即使核心线程最初只是在新任务到达时才创建和启动的，也可以使用方法prestartCoreThread()或 prestartAllCoreThreads()对其进行动态重写。如果构造带有非空队列的池，则可能希望预先启动线程。

创建新线程Creating new threads

使用ThreadFactory创建新线程。如果没有另外说明，则在同一个ThreadGroup中一律使用Executors.defaultThreadFactory() 创建线程，并且这些线程具有相同的NORM_PRIORITY优先级和非守护进程状态。通过提供不同的ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，等等。如果从newThread返回null时ThreadFactory未能创建线程，则执行程序将继续运行，但不能执行任何任务。

保持活动时间 Keep-alive times

如果池中当前有多于corePoolSize的线程，则这些多出的线程在空闲时间超过keepAliveTime时将会终止（参见 getKeepAliveTime(java.util.concurrent.TimeUnit)）。这提供了当池处于非活动状态时减少资源消耗的方法。如果池后来变得更为活动，则可以创建新的线程。也可以使用方法setKeepAliveTime(long, java.util.concurrent.TimeUnit)动态地更改此参数。使用Long.MAX_VALUE TimeUnit.NANOSECONDS的值在关闭前有效地从以前的终止状态禁用空闲线程。默认情况下，保持活动策略只在有多于corePoolSizeThreads的线程时应用。但是只要keepAliveTime值非0，allowCoreThreadTimeOut(boolean)方法也可将此超时策略应用于核心线程。

排队Queuing

    所有BlockingQueue都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互：
        如果运行的线程少于corePoolSize，则Executor始终首选添加新的线程，而不进行排队。
        如果运行的线程等于或多于corePoolSize，则Executor始终首选将请求加入队列，而不添加新的线程。
        如果无法将请求加入队列，则创建新的线程，除非创建此线程超出maximumPoolSize，在这种情况下，任务将被拒绝。

    排队有三种通用策略There are three general strategies for queuing:
        1 直接提交 Direct handoffs。工作队列的默认选项是SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界maximumPoolSizes以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
        2 无界队列 Unbounded queues。使用无界队列（例如，不具有预定义容量的LinkedBlockingQueue）将导致在所有 corePoolSize线程都忙时新任务在队列中等待。这样，创建的线程就不会超过corePoolSize。（因此，maximumPoolSize的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在Web页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
        3 有界队列Bounded queues。当使用有限的maximumPoolSizes时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低CPU使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是I/O边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

被拒绝的任务 Rejected tasks

    当Executor已经关闭，并且Executor将有限边界用于最大线程和工作队列容量，且已经饱和时，在方法 execute(java.lang.Runnable) 中提交的新任务将被拒绝。在以上两种情况下， execute方法都将调用其 RejectedExecutionHandler的RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。
    下面提供了四种预定义的处理程序策略：
        在默认的ThreadPoolExecutor.AbortPolicy中，处理程序遭到拒绝将抛出运行时RejectedExecutionException。
        在ThreadPoolExecutor.CallerRunsPolicy中，线程调用运行该任务的execute本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。
        在ThreadPoolExecutor.DiscardPolicy中，不能执行的任务将被删除。
        在ThreadPoolExecutor.DiscardOldestPolicy中，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。
    定义和使用其他种类的RejectedExecutionHandler类也是可能的，但这样做需要非常小心，尤其是当策略仅用于特定容量或排队策略时。

钩子(hook)方法 Hook methods

此类提供protected可重写的beforeExecute(java.lang.Thread, java.lang.Runnable)和afterExecute(java.lang.Runnable, java.lang.Throwable)方法，这两种方法分别在执行每个任务之前和之后调用。它们可用于操纵执行环境；例如，重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外，还可以重写方法terminated()来执行Executor完全终止后需要完成的所有特殊处理。
如果钩子 (hook) 或回调方法抛出异常，则内部辅助线程将依次失败并突然终止。

队列维护 Queue maintenance

方法getQueue()允许出于监控和调试目的而访问工作队列。强烈反对出于其他任何目的而使用此方法。 remove(java.lang.Runnable) 和 purge() 这两种方法可用于在取消大量已排队任务时帮助进行存储回收。

终止 Finalization

程序 AND 不再引用的池没有剩余线程会自动 shutdown。如果希望确保回收取消引用的池（即使用户忘记调用 shutdown()），则必须安排未使用的线程最终终止：设置适当保持活动时间，使用 0 核心线程的下边界和/或设置 allowCoreThreadTimeOut(boolean)。

扩展ThreadPoolExecutor示例

此类的大多数扩展可以重写一个或多个受保护的钩子 (hook) 方法。例如，下面是一个添加了简单的暂停/恢复功能的子类：
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();
     }
   }
}

Callable、Future和CompletionService

在整理执行器Executor之前，需要提到这样几个类/接口，这些类在使用执行器实现并发开发的时候是比较有用的，他们就是Callable、Future和CompletionService。

0. Callable

在最开始整理Java多线程并发的文章时就提到过了，Runnable是在JDK1.0中就有的。我们再来回头看下java.lang.Runnable:

public 

interface Runnable {
   publicabstractvoidrun();
}

可以看到其只有一个public void修饰的run()方法，注意是返回类型为void方法，这就意味着对这个方法的调用不会有有返回结果的，但很多场景一个任务完成执行，我们是需要知道它的结果的。为了弥补这个不足，从JDK1.5开始，在java.util.concurrent包中就有了Callable这个接口。

public 

interface Callable<V> {
    V call()throwsException;
}

注意到，其中call()方法除了有返回结果以外，比起run()还有异常抛出，这个使用时是要注意的。

在JavaSE5之后，执行器Executor是JDK提供给我们的良好工具，在ExecutorService中也有了支持Callable的submit()方法，那么对于其call()的执行结果我们如何来取呢，这就要提到另一个类——java.util.concurrent.Future。

1. Future

此前整理Java并发文章，在写任务终止的时候就提到过这个接口，我们来回顾一下：

public 

interface Future<V> {
    booleancancel(booleanmayInterruptIfRunning);
    booleanisCancelled();
    booleanisDone();
    V get()throwsInterruptedException, ExecutionException;
    V get(longtimeout, TimeUnit unit)
        throwsInterruptedException, ExecutionException, TimeoutException;
}

接口中列出了5个方法，分两大方面。除了对任务终止处理以外，Future的作用就是通过get()方法拿到特定类型的结果。

2. FutureTask类与RunnableFuture接口

和Future接口一起，在java.util.concurrent包中，FutureTask是其具体的实现类。而在JDK1.6中，又引入了RunnableFuture这样一个概念:

public 

interface RunnableFuture extendsRunnable, Future {
    voidrun();
}

乍一看，其实并没有什么，不过是把Runnable和Future两个接口捏到一起了。实际上，这个接口用意应该是这样的，将需要run()的任务和结果结合在一起，执行了run()能够保证结果被设置从而可以获取到。

在ExecutorService中，submit()方法会有很多重载实现，有的用Runnable参数，有的用Callable参数。而对于submit()方法本身的实际操作，就是：

执行任务
返回Future对象

在实现中，AbstractExecutorService的submit()方法无论传入的是Callable还是Runnable，都会调用newTaskFor()将其转变为一个RunnableFuture对象，这个对象既会被用来调用Executor的execute()方法，也会作为Future结果返回。

JDK1.6之后，FutureTask也就成为了RunnableFuture的一个实现，当然也还是Future的实现类。我们再来简单看下它的实现。

前面文章提到过AbstractQueuedSynchronizer类(AQS)的应用，实际上FutureTask中也有一个名为Sync而且继承于AQS的内部类。在FutureTask的实现中，每个任务都有执行状态，其巧妙地运用了AQS提供的state属性和protected方法，保证了对Future结果获取线程的阻塞和唤醒。

3. CompleteService

最后稍微提一下CompleteService接口，这个接口是为了方便多个任务执行时，可以方便得获取到执行任务的Future结果。接口内容如下：

public 

interface CompletionService<V> {
    Future<V> submit(Callable<V> task);
    Future<V> submit(Runnable task, V result);
    Future<V> take()throwsInterruptedException;
    Future<V> poll();
    Future<V> poll(longtimeout, TimeUnit unit)throwsInterruptedException;
}

同样，也是五个方法，分为两大方面。一个是对Callable和Runnable类型参数的任务提交，另一方面则是尝试对结果以不同的方式进行获取，take()方法一般是阻塞式的获取，后两者则更灵活。

通常来讲，CompleteService是要和Executor结合在一起使用的，这在后面讲Executor的时候还会详细讲到。

任务执行器Executor/ExecutorService和ExecutorCompletionService

上篇文章讲到了Callable和Futre，这篇文章接着来说说执行器相关的接口和类。在JavaSE5之后，在API的java.util.concurrent包中给出了Executor这样一个接口。因为在Java7之前，JDK中这个接口最终都是以一个线程池ThreadPool的方式实现的，所以很多人也直接叫“线程池”。除了ThreadPool之外，还有ForkJoinPool实现，本文中，我们不妨直接翻译Executor为执行器。

Executor本身只是个接口，但和这个概念相关有很多类：包括Executor本身、ExecutorService、AbstractExecutorService、提供诸多静态方法的工具类Executors和相关的ExecutorCompletionService类，下面会对这些接口和类一一整理。除此之外，还有刚刚提到的ThreadPoolExecutor和ForkJoinPool，这两个类后面我们再说。

0. 综述和类层次结构

JavaSE5之后的任务执行器Executor实际上取代了Java并发开发中直接使用Thread对象的操作，以Command Pattern（命令模式）的设计模式实现，成为Java并发开发的主角。

在eclipse中的Type Hierarchy View中可以看到执行器类的层次结构如下：

Executor是最为抽象的接口，ExecutorService继而扩展了Executor， AbstractExecutorService 则给出了抽象的实现。

1. Executor

先看java.util.concurrent.Executor这个接口：

public 

interface Executor {
    voidexecute(Runnable command);
}

同Runnable和Callable一样简洁，只有一个方法。注意execute()这个方法只允许有一个Runnable参数，也就意味着想用前文中的Callable，需要做适配工作。

2. ExecutorService

java.util.concurrent.ExecutorService也是一个接口，扩展了Executor，提供了更为丰富的方法。

public 

interface ExecutorService extendsExecutor {
    voidshutdown();
    List shutdownNow();
    booleanisShutdown();
    booleanisTerminated();
    booleanawaitTermination(longtimeout, TimeUnit unit)
        throwsInterruptedException;
     Future submit(Callable task);
     Future submit(Runnable task, T result);
    Future<?> submit(Runnable task);
     List<Future> invokeAll(Collection<?extendsCallable> tasks)
        throwsInterruptedException;
     List<Future> invokeAll(Collection<?extendsCallable> tasks,
                                  longtimeout, TimeUnit unit)
        throwsInterruptedException;
     T invokeAny(Collection<?extendsCallable> tasks)
        throwsInterruptedException, ExecutionException;
     T invokeAny(Collection<?extendsCallable> tasks,
                    longtimeout, TimeUnit unit)
        throwsInterruptedException, ExecutionException, TimeoutException;
}

一共12个方法，其中一部分是和执行器生命周期相关的方法，而另一部分则是以各种方式提交要执行的任务的方法。像submit()就是提交任务的一个方法，在实现中做了适配的工作，无论参数是Runnable还是Callable，执行器都会正确执行。当然，这实际上用到的是前文提过的RunnableFuture的实现类FutureTask。

3. AbstractExecutorService

java.util.concurrent.AbstractExecutorService这个类是ExecutorService的一个抽象实现。其中，提交任务的各类方法已经给出了十分完整的实现。之所以抽象，是因为和执行器本身生命周期相关的方法，在此类中并未给出任何实现，需要子类扩展完善。

4. Executors

这个类同Arrays和Collections很类似，java.util.concurrent.Executors是个工具类，提供了很多静态的工具方法。其中很多对于执行器来说就是初始化构建用的工厂方法。其中部分方法名整理成如下列表：

重载实现的newFixedThreadPool()
重载实现的newSingleThreadExecutor()
重载实现的newCachedThreadPool()
重载实现的newSingleThreadScheduledExecutor()
重载实现的newScheduledThreadPool()

这些方法返回的ExecutorService对象最终都是由ThreadPoolExecutor实现的，根据不同的需求以不同的参数配置，或经过其它类包装。其中，Executors中的一些内部类就是用来做包装用的。

Executors还提供了这样两个静态重载的方法:

public 

static ExecutorService unconfigurableExecutorService(ExecutorService executor);
public 

static ScheduledExecutorService unconfigurableScheduledExecutorService(ScheduledExecutorService executor);

这两个方法对参数中的executor做了包装，保证了Executor在初始化构造后不再可配置。

此外，Executors类中还有静态的defaultThreadFactory()方法，免去了自己构造ThreadFactory或是需要线程时对Thread类的new操作。

关于更多Executors类中工具方法的使用，还是建议去看API JavaDoc和JDK源码。

5. ExecutorCompletionService

前一篇文章提过了CompletionService这个接口，和执行器相关的，java.util.concurrent中提供了这个接口的一个实现，就是ExecutorCompletionService。

这个实现类主要做的就是将执行完成的任务结果放到阻塞队列中，这样等待结果的线程，如执行take()方法会得到结果并恢复执行。

ExecutorCompletionService有3个属性：

AbstractExecutorService类的对象aes
Executor类的对象executor
BlockingQueue<Future<V>>的completionQueue

通常，如果executor是AbstractExecutorService的一个实现，则将其赋值给aes属性，否则赋值为null。

在这个类中，executor负责执行任务，而aes则负责做适配处理，返回包装好任务的FutureTask对象。

这里面有一个对于实现功能很重要的内部类QueueingFuture，实现如下：

private 

class QueueingFuture extendsFutureTask {
    QueueingFuture(RunnableFuture task) {
        super(task,null);
        this.task = task;
    }
    protectedvoiddone() { completionQueue.add(task); }
    privatefinalFuture task;
}

主要是扩展了FutureTask的done方法，将执行结果放入BlockingQueue中。

ThreadPoolExecutor的应用和实现分析（上）—— 应用

前一篇文章写道了Executors类，其中提供了几个构造Executor的工厂方法。但在实现上，这些执行器最终都是采用了java.util.concurrent.ThreadPoolExecutor类的对象。接下来的文章我们就来了解和分析一下ThreadPoolExecutor这个类。由于篇幅比较长，这一篇主要从应用的角度，对ThreadPoolExecutor的使用做简单整理，对于任务提交和生命周期管理等源码实现的分析会在后面文章逐步整理。

0. ThreadPoolExecutor的初步认识

在之前整理Java并发开发的文章中提到，在JavaSE5之后，直接使用Thread类并不是被提倡的并发开发方式，而Executor成为主要角色。

这其中有如下几个原因：

复用效率。Thread每一次构建和销毁都有一定成本，而线程池执行器中，线程是可以复用的，这在一定程度上降低了线程的维护成本，提高了开发和运行效率。
资源限制和管理。前面整理并发的文章中我们也提到了，启用多线程并发开发实际上是为了更充分的利用处理器资源，但这并不意味着开启的线程越多越好。每个线程都占有一定的系统资源，如果为了处理某项任务，对开启线程不加限制，那么最终会使系统资源耗尽导致异常问题发生。因此，对于线程的启用需要有一个限制和管理，而线程池执行器将这些方面融入，降低了并发多线程的管理成本。
数据统计维护。除了上述两点以外，对于多线程执行中的一些数据，线程池执行器也进行了维护，方便了并发开发。

Executors这个工具类中提供了如下静态方法：

Executors.newCachedThreadPool() 无限大小的线程池，线程会自动重用
Executors.newFixedThreadPool(int) 固定线程数的线程池
Executors.newSingleThreadExecutor() 单线程执行器

而实际上他们是调用了ThreadPoolExecutor类的构造方法来创建Executor对象的。java.util.concurrent.ThreadPoolExecutor全参数构造方法如下:

public 

ThreadPoolExecutor(int 

corePoolSize,
                              intmaximumPoolSize,
                              longkeepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

ThreadPoolExecutor类一共有4个重载的构造方法，但最终调用的都是这个。这个构造方法的参数翠泉，一共提供了7个参数，这在Java API中的方法中，已经算是比较多的了。7个参数的简要说明如下：

corePoolSize 是线程池的核心线程数，通常线程池会维持这个线程数
maximumPoolSize 是线程池所能维持的最大线程数
keepAliveTime 和 unit 则分别是超额线程的空闲存活时间数和时间单位
workQueue 是提交任务到线程池的入队列
threadFactory 是线程池创建新线程的线程构造器
handler 是当线程池不能接受提交任务的时候的处理策略

更详细的解释下面会分点逐一说明。

1. 线程创建和任务提交的条件和步骤

当一个新任务被提交给ThreadPoolExecutor的时候，处理流程大概是这样的:

如果当前线程池中线程的数目低于corePoolSize，则创建新线程执行提交的任务，而无需检查当前是否有空闲线程
如果提交任务时线程池中线程数已达到corePoolSize，则将提交的任务加入等待执行队列
如果提交任务时等待执行的任务队列是有限队列，而且已满，则在线程池中开辟新线程执行此任务
如果线程池中线程数目已达到maximumPoolSize，则提交的任务交由RejectedExecutionHandler处理

默认情况下，ThreadPoolExecutor的线程数是根据需求来延迟初始化的，即有新任务加进来的时候才会挨个创建线程。

除此之外，线程池执行器也提供了提前创建初始化线程的方法：

public boolean prestartCoreThread()
public int prestartAllCoreThreads()

分别是预先创建一个线程和预先创建线程直到线程数到达核心线程数corePoolSize。

2. 线程数目的维护

刚刚提到，ThreadPoolExecutor有corePoolSize和maximum两个变量来维护线程池中的线程个数，提交任务的时候会有线程数目的增长，那线程的个数又是怎么来维护的呢。构造方法里还有两个参数，分别是keepAlive和unit，这两个参数确定了一个时间间隔，也就是空闲线程存活的时间间隔。默认情况下，当线程池中的线程个数超出了corePoolSize，那么空闲的线程一旦超出额定的存活时间就会被终止，以节省系统资源。在JDK1.6之后，增加了allowsCoreThreadTimeOut这个boolean属性和读写属性的方法，用来标志核心线程空闲超时是否也可以终止掉。

3. 线程入队列和任务丢弃原则简述

除了前面描述涉及到的四个属性和ThreadFactory之外，还有两个分别是workQueue和handler，分别是BlockingQueue和RejectedExecutionHandler类型。

BlockingQueue只是一个接口，它所表达的是当队列为空或者已满的时候，需要阻塞以等待生产者/消费者协同操作并唤醒线程。其有很多不同的具体实现类，各有特点。有的可以规定队列的长度，也有一些则是无界的。

按照Executors类中的几个工厂方法，分别使用的是：

LinkedBlockingQueue。CachedThreadPool使用的是这个BlockingQueue，队列长度是无界的，适合用于提交任务相互独立无依赖的场景。
SynchronousQueue。 FixedThreadPool和SingleThreadExecutor使用的是这个BlockingQueue，通常要求线程池不设定最大的线程数，以保证提交的任务有机会执行而不被丢掉。通常这个适合任务间有依赖的场景。

当然，开发者也可以定制ThreadPoolExecutor时使用ArrayBlockingQueue有界队列。

对于任务丢弃，ThreadPoolExecutor以内部类的形式实现了4个策略。分别是：

CallerRunsPolicy。提交任务的线程自己负责执行这个任务。
AbortPolicy。使Executor抛出异常，通过异常做处理。
DiscardPolicy。丢弃提交的任务。
DiscardOldestPolicy。丢弃掉队列中最早加入的任务。

在调用构造方法时，参数中未指定RejectedExecutionHandler情况下，默认采用AbortPolicy。

ThreadPoolExecutor的应用和实现分析（中）—— 任务处理相关源码分析

前面一篇文章从Executors中的工厂方法入手，已经对ThreadPoolExecutor的构造和使用做了一些整理。而这篇文章，我们将接着前面的介绍，从源码实现上对ThreadPoolExecutor在任务的提交、执行，线程重用和线程数维护等方面做下分析。

0. ThreadPoolExecutor类的声明属性变量分析

public 

class ThreadPoolExecutor extendsAbstractExecutorService

从这个类声明中我们可以看到java.util.ThreadPoolExecutor是继承于AbstractExecutorService的，而之前的文章我也提到过，AbstractExecutorService已经实现了一些任务提交处理的方法，如submit()方法都是在这个抽象类中实现的。但submit()方法，最后也是会调用ThreadPoolExecutor的execute()方法。

打开SunJDK中的ThreadPoolExecutor类源码，除了上篇文章提到的一些和构造方法中参数对应的属性之外，让我们看看还有什么：

mainLock 对整个ThreadPoolExecutor对象的锁
workers 存储工作线程对应Worker对象的HashSet
termination 线程池ThreadPoolExecutor对象的生命周期终止条件，和mainLock相关
largestPoolSize 线程池跑过的最大线程数
completedTaskCount 完成任务数
ctl 执行器ThreadPoolExecutor的生命周期状态和活动状态的worker数封装

稍微需要说一下最后一个, ctl是一个AtomicInteger对象，以位运算的方式打包封装了当前线程池ThreadPoolExecutor对象的状态和活动线程数两个数据

1. 执行器状态

ExecutorService中已经指定了这个接口对应的类要实现的方法，其中就包括shutdown()和shutdownNow()等方法。在ThreadPoolExecutor中指明了状态的含义，并包含其于ctl属性中。

ThreadPoolExecutor对象有五种状态，如下:

RUNNING 在ThreadPoolExecutor被实例化的时候就是这个状态
SHUTDOWN 通常是已经执行过shutdown()方法，不再接受新任务，等待线程池中和队列中任务完成
STOP 通常是已经执行过shutdownNow()方法，不接受新任务，队列中的任务也不再执行，并尝试终止线程池中的线程
TIDYING 线程池为空，就会到达这个状态，执行terminated()方法
TERMINATED terminated()执行完毕，就会到达这个状态，ThreadPoolExecutor终结

2. Worker内部类

它既实现了Runnable，同时也是一个AQS ( AbstractQueuedSynchronizer )。

private 

final class Worker
extends 

AbstractQueuedSynchronizer
implementsRunnable

封装了3样东西，Runnable类的首个任务对象，执行的线程thread和完成的任务数(volatile)completedTasks。

final 

Thread thread;
Runnable firstTask;
volatile 

long completedTasks;

这个类还提供了interruptIfStarted()这样一个方法，里面做了（getState()>= 0）的判断。与此呼应，Worker的构造方法里对state设置了-1，避免在线程执行前被停掉。

Worker(Runnable firstTask) {
    setState(-1);// inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

3. 提交任务

上篇文章已经提到了，提交新任务的时候，如果没达到核心线程数corePoolSize，则开辟新线程执行。如果达到核心线程数corePoolSize, 而队列未满，则放入队列，否则开新线程处理任务，直到maximumPoolSize，超出则丢弃处理。

这段源码逻辑如下，不细说了。

public 

void execute(Runnable command) {
    if(command ==null)
        thrownewNullPointerException();
 
    intc = ctl.get();
    if(workerCountOf(c) < corePoolSize) {
        if(addWorker(command,true))
            return;
        c = ctl.get();
    }
    if(isRunning(c) && workQueue.offer(command)) {
        intrecheck = ctl.get();
        if(! isRunning(recheck) && remove(command))
            reject(command);
        elseif(workerCountOf(recheck) ==0)
            addWorker(null,false);
    }
    elseif(!addWorker(command,false))
        reject(command);
}

4. addWorker()的实现

在上面提交任务的时候，会出现开辟新的线程来执行，这会调用addWorker()方法。

private 

boolean addWorker(Runnable firstTask, booleancore) {
    retry:
    for(;;) {
        intc = ctl.get();
        intrs = runStateOf(c);
 
        // Check if queue empty only if necessary.
        if(rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask ==null&&
               ! workQueue.isEmpty()))
            returnfalse;
 
        for(;;) {
            intwc = workerCountOf(c);
            if(wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                returnfalse;
            if(compareAndIncrementWorkerCount(c))
                breakretry;
            c = ctl.get(); // Re-read ctl
            if(runStateOf(c) != rs)
                continueretry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
 
    booleanworkerStarted =false;
    booleanworkerAdded =false;
    Worker w =null;
    try{
        finalReentrantLock mainLock =this.mainLock;
        w =newWorker(firstTask);
        finalThread t = w.thread;
        if(t !=null) {
            mainLock.lock();
            try{
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                intc = ctl.get();
                intrs = runStateOf(c);
 
                if(rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask ==null)) {
                    if(t.isAlive())// precheck that t is startable
                        thrownewIllegalThreadStateException();
                    workers.add(w);
                    ints = workers.size();
                    if(s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded =true;
                }
            }finally{
                mainLock.unlock();
            }
            if(workerAdded) {
                t.start();
                workerStarted =true;
            }
        }
    }finally{
        if(! workerStarted)
            addWorkerFailed(w);
    }
    returnworkerStarted;
}

代码较长，我们可以分两大部分看：

第一段从第3行到第26行，是双层无限循环，尝试增加线程数到ctl变量，并且做一些比较判断，如果超出线程数限定或者ThreadPoolExecutor的状态不符合要求，则直接返回false，增加worker失败。

第二段从第28行开始到结尾，把firstTask这个Runnable对象传给Worker构造方法，赋值给Worker对象的task属性。Worker对象把自身（也是一个Runnable）封装成一个Thread对象赋予Worker对象的thread属性。锁住整个线程池并实际增加worker到workers的HashSet对象当中。成功增加后开始执行t.start()，就是worker的thread属性开始运行，实际上就是运行Worker对象的run方法。Worker的run()方法实际上调用了ThreadPoolExecutor的runWorker()方法。

5. 任务的执行runWorker()

final 

void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask =null;
        w.unlock();// allow interrupts
        booleancompletedAbruptly =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;throwx;
                    }catch(Error x) {
                        thrown = x;throwx;
                    }catch(Throwable x) {
                        thrown = x;thrownewError(x);
                    }finally{
                        afterExecute(task, thrown);
                    }
                }finally{
                    task =null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly =false;
        }finally{
            processWorkerExit(w, completedAbruptly);
        }
    }

这段代码实际上就是执行提交给线程池执行的Runnable任务的实际内容。其中，值得注意的有以下几点：

线程开始执行前，需要对worker加锁，完成一个任务后执行unlock()
在任务执行前后，执行beforeExecute()和afterExecute()方法
记录任务执行中的异常后，继续抛出
每个任务完成后，会记录当前线程完成的任务数
当worker执行完一个任务的时候，包括初始任务firstTask，会调用getTask()继续获取任务，这个方法调用是可以阻塞的
线程退出，执行processWorkerExit(w, completedAbruptly)处理

5. Worker线程的复用和任务的获取getTask()

在上一段代码中，也就是runWorker()方法，任务的执行过程是嵌套在while循环语句块中的。每当一个任务执行完毕，会从头开始做下一次循环执行，实现了空闲线程的复用。而要执行的任务则是来自于getTask()方法：

private 

Runnable getTask() {
        booleantimedOut =false;//

 Did the last poll() time out?
 
        retry:
        for(;;) {
            intc = ctl.get();
            intrs = runStateOf(c);
 
            // Check if queue empty only if necessary.
            if(rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                returnnull;
            }
 
            booleantimed;     // Are workers subject to culling?
 
            for(;;) {
                intwc = workerCountOf(c);
                timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
                if(wc <= maximumPoolSize && ! (timedOut && timed))
                     break;
                if(compareAndDecrementWorkerCount(c))
                     returnnull;
                c = ctl.get();
                // Re-read ctl
                if(runStateOf(c) != rs)
                     continueretry;
                // else CAS failed due to workerCount change; retry inner loop
             }
             try{
                 Runnable r = timed ?
                     workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                     workQueue.take();
                 if(r !=null)
                     returnr;
                 timedOut =true;
             }catch(InterruptedException retry) {
                 timedOut =false;
             }
         }
     }

getTask()实际上是从工作队列（workQueue）中取提交进来的任务。这个workQueue是一个BlockingQueue，通常当队列中没有新任务的时候，则getTask()会阻塞。另外，还有定时阻塞这样一段逻辑：如果从队列中取任务是计时的，则用poll()方法，并设置等待时间为keepAlive，否则调用阻塞方法take()。当poll()超时，则获取到的任务为null，timeOut设置为 true。这段代码也是放在一个for(;;)循环中，前面有判断超时的语句，如果超时，则return
null。这意味着runWorker()方法的while循环结束，线程将退出，执行processWorkerExit()方法。

回头看看是否计时是如何确定的。

int 

wc = workerCountOf(c);
timed = allowCoreThreadTimeOut || wc &gt; corePoolSize;

即判断当前线程池的线程数是否超出corePoolSize，如果超出这个值并且空闲时间多于keepAlive则当前线程退出。

另外一种情况就是allowCoreThreadTimeOut为true，就是允许核心在空闲超时的情况下停掉。

6. 线程池线程数的维护和线程的退出处理

刚刚也提到了，我们再看下processWorkerExit()方法。这个方法最主要就是从workers的Set中remove掉一个多余的线程。

private 

void processWorkerExit(Worker w, booleancompletedAbruptly) {
         if(completedAbruptly)// If abrupt, then workerCount wasn't adjusted
             decrementWorkerCount();
         finalReentrantLock mainLock =this.mainLock;
         mainLock.lock();
         try{
             completedTaskCount += w.completedTasks;
             workers.remove(w);
         }finally{
             mainLock.unlock();
         }
         tryTerminate();
         intc = ctl.get();
         if(runStateLessThan(c, STOP)) {
             if(!completedAbruptly) {
                 intmin = allowCoreThreadTimeOut ?0: corePoolSize;
                 if(min ==0

&& ! workQueue.isEmpty())
                    min =1;
                 if(workerCountOf(c) >= min)
                    return;// replacement not needed
            }
            addWorker(null,false);
        }
    }

这个方法的第二个参数是判断是否在runWorker()中正常退出了循环向下执行，如果不是，说明在执行任务的过程中出现了异常，completedAbruptly为true，线程直接退出，需要直接对活动线程数减1 。

之后，加锁统计完成的任务数，并从workers这个集合中移除当前worker。

执行tryTerminate()，这个方法后面会详细说，主要就是尝试将线程池推向TERMINATED状态。

最后比较当前线程数是不是已经低于应有的线程数，如果这个情况发生，则添加无任务的空Worker到线程池中待命。

以上，增加新的线程和剔除多余的线程的过程大概就是如此，这样线程池能保持额定的线程数，并弹性伸缩，保证系统的资源不至于过度消耗。

ThreadPoolExecutor的应用和实现分析（下）—— 生命周期相关源码分析

前面提到ExecutorService是对Executor的扩展，增加了submit()等提交和调用任务的方法，也增加了对Executor生命周期的定义。本篇将结合shutdown()、shutdownNow()、awaitTermination()和tryTerminate()几个方法的实现对ThreadPoolExecutor生命周期相关的逻辑做分析整理。

0. ExecutorService的生命周期方法和ThreadPoolExecutor的生命周期实现

ExecutorService中，和生命周期相关的，声明了5个方法：

awaitTermination() 阻塞等待shutdown请求后所有线程终止，会有时间参数，超时和中断也会令方法调用结束
isShutdown() 通过ctl属性判断当前的状态是否不是RUNNING状态
isTerminated() 通过ctl属性判断当前的状态是否为TERMINATED状态
shutdown() 关闭Executor，不再接受提交任务
shutdownNow() 关闭Executor，不再接受提交任务，并且不再执行入队列中的任务

前面的文章也已经介绍过，在ThreadPoolExecutor的实现中，定义了五个生命周期状态，标识了整个线程池对象所处的状态阶段，用来实现生命周期相关的方法。

1. ThreadPoolExecutor的shutdown()

我们先来看看shutdown()方法

public 

void shutdown() {
    finalReentrantLock mainLock =

this.mainLock;
    mainLock.lock();
    try{
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown();// hook for ScheduledThreadPoolExecutor
    }finally

{
        mainLock.unlock();
    }
    tryTerminate();
}

其实逻辑比较简单，尝试将状态切换到SHUTDOWN，这样就不会再接收新的任务提交。对空闲线程进行中断调用。最后检查线程池线程是否为0，并尝试切换到TERMINATED状态。

2. ThreadPoolExecutor的shutdownNow()

再来看看shutdownNow()方法

public 

List shutdownNow() {
    List tasks;
    finalReentrantLock mainLock =

this.mainLock;
    mainLock.lock();
    try{
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    }finally

{
        mainLock.unlock();
    }
    tryTerminate();
    returntasks;
}

主要所做的事情就是切换ThreadPoolExecutor到STOP状态，中断所有worker，并将任务队列中的任务取出来，不再执行。最后尝试修改状态到TERMINATED。

3. shutdown()和shutdownNow()的区别

shutdown()新的任务不会再被提交到线程池，但之前的都会依旧执行，通过中断方式停止空闲的（根据没有获取锁来确定）线程。

shutdownNow()则向所有正在执行的线程发出中断信号以尝试终止线程，并将工作队列中的任务以列表方式的结果返回。

两者区别：

是一个要将线程池推到SHUTDOWN状态，一个将推到STOP状态
并且对运行的线程处理方式不同，shutdown()只中断空闲线程，而shutdownNow()会尝试中断所有活动线程
还有就是对队列中的任务处理，shutdown()队列中已有任务会继续执行，而shutdownNow()会直接取出不被执行

相同的是都在最后尝试将线程池推到TERMINATED状态。

4. ThreadPoolExecutor的awaitTermination()

阻塞等待shutdown请求后所有线程终止，会有时间参数，超时和中断也会令方法调用结束。

public 

boolean awaitTermination(longtimeout, TimeUnit unit)
    throwsInterruptedException {
    longnanos = unit.toNanos(timeout);
    finalReentrantLock mainLock =

this.mainLock;
    mainLock.lock();
    try{
        for(;;) {
            if(runStateAtLeast(ctl.get(), TERMINATED))
                returntrue;
            if(nanos <=

0)
                returnfalse;
            nanos = termination.awaitNanos(nanos);
        }
    }finally

{
        mainLock.unlock();
    }
}

实际所做的就是Condition的定时await调用。用于状态依赖的线程阻塞。

5. tryTerminate()

tryTerminate()的意义就在于尝试进入终止状态，当ctl中worker数字为0时执行terminated()方法，否则等锁中断一个空闲的Worker。

final 

void tryTerminate() {
    for(;;) {
        intc = 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;
        }
 
        finalReentrantLock 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
    }
}

其中interruptIdleWorkers()方法这里就不列代码了，空闲的worker主要是通过worker的tryLock()来确认的，因为执行任务的worker互斥地锁定对象。

中断worker导致线程退出，最终还会循环尝试终止其它的空闲线程，直到整个ThreadPoolExecutor最后终结。

6. ThreadPoolExecutor生命周期的扩展点

在生命周期上，ThreadPoolExecutor为扩展的类提供了一些扩展点，这是很好的设计，对扩展开放。

其中声明了如下protected的方法：

beforeExecute() 在每个任务执行前做的处理
afterExecute() 在每个任务执行后做的处理
terminated() 在ThreadPoolExecutor到达TERMINATED状态前所做的处理
finalize() 有默认实现，直接调用shutdown()，以保证线程池对象回收
onShutdown() 在shutdown()方法执行到最后时调用，在java.util.concurrent.ScheduledThreadPoolExecutor类实现中用到了这个扩展点，做一些任务队列的清理操作。

下篇Java并发的文章会再介绍线程池饱和时的丢弃处理，更多关于ThreadPoolExecutor的分析和扩展还望大家整理指教。

ThreadPoolExecutor的应用和实现分析（续）—— 任务饱和丢弃策略

在前面三篇文章中，我们已经对ThreadPoolExecutor的应用以及任务处理和生命周期相关的源码实现做了整理分析。这篇我们简要整理下java.util.concurrent包中的RejectedExecutionHandler这个接口和对应的实现类。

0. RejectedExecutionHandler接口

当ThreadPoolExecutor执行任务的时候，如果线程池的线程已经饱和，并且任务队列也已满。那么就会做丢弃处理，这也是execute()方法实现中的操作，源码如下:

else 

if (!addWorker(command, false))
        reject(command);

这个reject()方法很简单，直接调用丢弃处理的handler方法的rejectedExecution()。

在java.util.concurrent中，专门为此定义了一个接口，是RejectedExecutionHandler：

public 

interface RejectedExecutionHandler {
    voidrejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

其中只有rejectedExecution()一个方法。返回为void，而参数一个是具体的Runnable任务，另一个则是被提交任务的ThreadPoolExecutor。

凡是实现了这个方法的类都可以作为丢弃处理器在ThreadPoolExecutor对象构造的时候作为参数传入，这个前面的文章已经提到过了。其中ThreadPoolExecutor给出了4种基本策略的实现。分别是:

CallerRunsPolicy
AbortPolicy
DiscardPolicy
DiscardOldestPolicy

下面分别详细说明。

1. 直接丢弃

这个也是实现最简单的类，其中的rejectedExecution()方法是空实现，即什么也不做，那么提交的任务将会被丢弃，而不做任何处理。

public 

static class DiscardPolicyimplements

RejectedExecutionHandler {
    publicDiscardPolicy() { }
 
    publicvoid

rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

这个策略使用的时候要小心，要明确需求。不然不知不觉的任务就丢了。

2. 丢弃最老

和上面的有些类似，也是会丢弃掉一个任务，但是是队列中最早的。

实现如下:

public 

static class DiscardOldestPolicyimplements

RejectedExecutionHandler {
    publicDiscardOldestPolicy() { }
 
    publicvoid

rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if(!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

注意，会先判断ThreadPoolExecutor对象是否已经进入SHUTDOWN以后的状态。之后取出队列头的任务并不做任何处理，即丢弃，再重新调用execute()方法提交新任务。

3. 废弃终止

这个RejectedExecutionHandler类和直接丢弃不同的是，不是默默地处理，而是抛出java.util.concurrent.RejectedExecutionException异常，这个异常是RuntimeException的子类。这个策略实现如下:

public 

static class AbortPolicyimplements

RejectedExecutionHandler {
    publicAbortPolicy() { }
 
    publicvoid

rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        thrownew

RejectedExecutionException("Task "+ r.toString() +
                                             " rejected from "+
                                             e.toString());
    }
}

注意，处理这个异常的线程是执行execute()的调用者线程。

4. 调用者执行策略

在这个策略实现中，任务还是会被执行，但线程池中不会开辟新线程，而是提交任务的线程来负责维护任务。

public 

void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if(!e.isShutdown()) {
        r.run();
    }
}

注意，和DiscardOldestPolicy同样，也会先判断ThreadPoolExecutor对象的状态，之后执行任务。这样处理的一个好处，是让caller线程运行任务，以推迟该线程进一步提交新任务，有效的缓解了线程池对象饱和的情况。

上面只是SunJDK中提供的4种最基本策略，开发者可以根据具体需求定制。

此外，前文提到ThreadPoolExecutor也可以进行扩展。在java.util.concurrent包中有ScheduledThreadPoolExecutor这样一个类，其扩展了ThreadPoolExecutor，实现了一些时间和任务调度相关的方法。这里我就不具体整理了。

BlockingQueue及其各个实现的分析整理

前面在整理的ThreadPoolExecutor类中，有一个从提交任务到线程池分配线程执行任务，有一个队列，而这个队列采用的就是BlockingQueue。BlockingQueue实际上定义了一个接口，在java.util.concurrent包中给出了这个接口的一些实现，这篇我们整理一下。

0. BlockingQueue简介

BlockingQueue是java.util.concurrent包中的接口，扩展了java.util中的的Queue接口。

在Java7的API中，这个接口有11个public方法。

但对于BlockingQueue来说，其本身就是一个就是一个阻塞队列，所以这些操作的方法中，最重要的两个就是put()和take()方法，这也是本篇中重点分析的地方，其它的方法可以参见JavaDoc文档。

BlockingQueue的实现有一个特点，队列元素不接受null值。

BlockingQueue这个接口在JDK中提供了很多具体实现，包括了数组、链表等实现，下面就对这些实现类简要分析下。

1. 数组实现的ArrayBlockingQueue

看下ArrayBlockingQueue的构造方法，一共有三个：

ArrayBlockingQueue(int capacity)
ArrayBlockingQueue(int capacity, boolean fair)
ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)

我们发现，构造方法中并没有无参的，这意味着队列的容量是没有默认的，在使用的时候需要给出容量值。

后两个构造方法还有fair这个参数。这个fair可谓是似曾相识，其实它就是ReentrantLock对象初始化要用到的那个参数。我们知道ArrayBlockingQueue既然是阻塞队列，那么一定会有阻塞和唤醒，这里的实现用到的是Condition的await()和signal() / signalAll()，而用Condition的前提就是有对应的Lock对象，在ArrayBlockingQueue实现中，take()和put()用的是统一的一个单锁。在ArrayBlockingQueue的某些并发操作方法中，是需要加锁来保证线程安全的，而这就是fair参数的作用。

对于队列“空”和“满”的情况，分别使用了两个Condition对象来维护。

另外，ArrayBlockingQueue类我们直接理解就是数组实现的阻塞队列。没错，其中的数据元素是用Object[]来保存的。对于take()和put()方法，则是分别使用了takeIndex和putIndex这两个索引值来记录存放数据的位置。

public 

E take() throws 

InterruptedException {
    finalReentrantLock lock =

this.lock;
    lock.lockInterruptibly();
    try{
        while(count ==

0)
            notEmpty.await();
        returnextract();
    }finally

{
        lock.unlock();
    }
}

如上，是take()方法实现的源码。逻辑很简单，先加锁，然后判断是否队列已空，如条件为真，则阻塞，然后取出队列中的元素。我们看到，阻塞是通过对notEmpty这个Condition对象的await()方法调用来做到的，与此对应，extract()方法中实际上也有一个notFull.signal()的调用。

2. 单向链表实现的LinkedBlockingQueue

LinkedBlockingQueue是JDK中BlockingQueue的有一个主要的实现。按照JavaDoc上所述，LinkedBlockingQueue是一个容量可选的阻塞队列。存在LinkedBlockingQueue()无参的默认构造方法实现，使用Integer.MAX_VALUE作为默认容量。

在LinkedBlockingQueue类的实现中，很重要的一个和ArrayBlockingQueue不同的地方，是对put()和take()分别使用了两个不同的锁，都使用了ReentrantLock实现。而针对“空”和“满”的阻塞条件，也是对这两个所对象分别构建的两个Condition对象（notEmpty和notFull），构成了双锁双条件。此外，LinkedBlockingQueue也为take和put操作分别维护了索引takeIndex和putIndex。两锁或者说队列状态的协调一致其实也是通过两个条件对象的await()和signal()来达成。

/** Lock held by take, poll, etc */
private 

final ReentrantLock takeLock = newReentrantLock();
 
/** Wait queue for waiting takes */
private 

final Condition notEmpty = takeLock.newCondition();
 
/** Lock held by put, offer, etc */
private 

final ReentrantLock putLock = newReentrantLock();
 
/** Wait queue for waiting puts */
privatefinal

Condition notFull = putLock.newCondition();

此外，对于队列中元素的计数，LinkedBlockingQueue也和ArrayBlockingQueue的实现略有不同，使用了AtomicInteger类对象。

对于put()和take()以及类似的操作，双锁避免了互相影响，一定意义上看，减小了操作的锁粒度，提高了并发性。

但对于其他操作，为了保证线程安全，都是双锁同时锁定。双锁使用要避免死锁问题，这个类实现中是统一定义了fullyLock()和fullyUnlock()的方法，先锁定的后释放，避免死锁发生的可能。

除了用数组和队列不同数据结构对BlockingQueue接口的基本实现外，还有其他几种有特殊功能的实现。

3. DelayQueue

基本特征是容量无界，实现上单锁单条件。

功能特点上，实际上是对优先级队列PriorityQueue类的一个封装。放入队列的元素要满足要求<E extends Delayed>。比较器是时间，因为：

public 

interface Delayed extendsComparable<Delayed>

元素需要给出getDelay()方法（实际上是Delayed接口的要求）。

等待第一个元素的线程被设置为leader，后续线程无限期等待，直到leader通知他们。队列中的数据元素超时后，元素可被返回。

4. 同步队列SynchronousQueue

这个类在Executors中使用ThreadPoolExecutor类构造CachedThreadPool的时候被用到了。SynchronousQueue的特点是，读取操作take()和放入操作put()同时完成才会同事解开阻塞。即一个元素只有当其本身被take()的时候put()才会被唤醒。没有容量的概念。

构造方法中可以带fair参数，分为公平和非公平实现，具体的实现分别为队列和栈，顺序不同。具体的代码实现依赖于内部类TransferQueue和TransferStack，逻辑较为复杂，这里不做细节分析。实现中的阻塞机制直接使用LockSupport的park()方法。

5. 顺便说说Exchanger类

这个类也是java.util.concurrent包中的，但和BlockingQueue并无直接层次结构关系。这里提到它主要是因为从用法上来看，相当于一个二项的SynchronousQueue。

具体实现上比较复杂，不做详细分析，记录下几点：

注意到Slot和Node都是AtomicReference，其compareAndSet并不是设置node或者item，而是引用值，巧妙的利用了Node的引用值和item做数据交换
（高并发情况）实现上用了类似concurrentHashMap的segment方式，有插槽Slot的概念
阻塞机制用Locksupport.park()

6. TransferQueue

最后说下TransferQueue这个接口，这个类是java.util.concurrent包中在Java7中增加的，可以看到注释中的“@since 1.7”。和前面的不同，TransferQueue只是一个接口，不是一个实现。在JDK1.7中，有LinkedTransferQueue这样一个实现类。需要注意区分，这个TransferQueue和SynchronousQueue的内部实现类TransferQueue不是同一个类。

这个接口/类实际上是一个比SynchronousQueue更灵活更高级的同步队列，放入新元素可以阻塞也可以非阻塞，并且也可以设定队列的元素容量。

这篇对BlockingQueue的小结就到这里。

参考

JDK6.0 ThreadPoolExecutor源码及注释

：http://www.molotang.com/articles/494.html

【上篇】Java concurrent Framework并发容器之concurrent.atomic包源码分析
【下篇】Java concurrent Framework并发之fork-join框架分析

作者: pyjama

该日志由 pyjama 于5年前发表在综合分类下，最后更新于 2019年09月28日.
转载请注明: Java concurrent Framework并发之Executor框架分析(核心的地方要多次研究) | 学步园 +复制链接

抱歉!评论已关闭.

学步园