Java并发编程实践 目录
并发编程 04—— 闭锁CountDownLatch 与 栅栏CyclicBarrier
并发编程 06—— CompletionService : Executor 和 BlockingQueue
并发编程 10—— 任务取消 之 关闭 ExecutorService
并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性
并发编程 13—— 线程池的使用 之 配置ThreadPoolExecutor 和 饱和策略
概述
第7部分 ScheduledThreadPoolExecutor
第1部分 线程池架构图
第2部分 Executor
它是"执行者"接口，它是来执行任务的。准确的说，Executor提供了execute()接口来执行已提交的 Runnable 任务的对象。Executor存在的目的是提供一种将"任务提交"与"任务如何运行"分离开来的机制。
它只包含一个函数接口：
void execute(Runnable command) 在未来某个时间执行给定的命令。
第3部分 ExecutorService
ExecutorService继承于Executor。它是"执行者服务"接口，它是为"执行者接口Executor"服务而存在的；准确的话，ExecutorService提供了"将任务提交给执行者的接口(submit方法)"，"让执行者执行任务(invokeAll, invokeAny方法)"的接口等等。
ExecutorService的函数列表
boolean awaitTermination(long timeout, TimeUnit unit) 请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行。 <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) 执行给定的任务，当所有任务完成时，返回保持任务状态和结果的 Future 列表。 <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) 执行给定的任务，当所有任务完成或超时期满时（无论哪个首先发生），返回保持任务状态和结果的 Future 列表。 <T> T invokeAny(Collection<? extends Callable<T>> tasks) 执行给定的任务，如果某个任务已成功完成（也就是未抛出异常），则返回其结果。 <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) 执行给定的任务，如果在给定的超时期满前某个任务已成功完成（也就是未抛出异常），则返回其结果。 boolean isShutdown() 如果此执行程序已关闭，则返回 true。 boolean isTerminated() 如果关闭后所有任务都已完成，则返回 true。 void shutdown() 启动一次顺序关闭，执行以前提交的任务，但不接受新任务。 List<Runnable> shutdownNow() 试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表。 <T> Future<T> submit(Callable<T> task) 提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future。 Future<?> submit(Runnable task) 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。 <T> Future<T> submit(Runnable task, T result) 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
第4部分 AbstractExecutorService
AbstractExecutorService是一个抽象类，它实现了ExecutorService接口。
AbstractExecutorService存在的目的是为ExecutorService中的函数接口提供了默认实现。
AbstractExecutorService函数列表
由于它的函数列表和ExecutorService一样，这里就不再重复列举了。
第5部分 ThreadPoolExecutor
ThreadPoolExecutor函数列表:
构造方法
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) 用给定的初始参数和默认的线程工厂及被拒绝的执行处理程序创建新的 ThreadPoolExecutor。 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) 用给定的初始参数和默认的线程工厂创建新的 ThreadPoolExecutor。 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) 用给定的初始参数和默认被拒绝的执行处理程序创建新的 ThreadPoolExecutor。 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) 用给定的初始参数创建新的 ThreadPoolExecutor。
ThreadPoolExecutor就是"线程池"。它继承于AbstractExecutorService抽象类。
线程池可以解决两个不同问题：由于减少了每个任务调用的开销，它们通常可以在执行大量异步任务时提供增强的性能，并且还可以提供绑定和管理资源（包括执行任务集时使用的线程）的方法。每个 ThreadPoolExecutor 还维护着一些基本的统计数据，如完成的任务数。
为了便于跨大量上下文使用，此类提供了很多可调整的参数和扩展钩子 (hook)。但是，强烈建议程序员使用较为方便的 Executors 工厂方法 Executors.newCachedThreadPool()（无界线程池，可以进行自动线程回收）、Executors.newFixedThreadPool(int)（固定大小线程池）和 Executors.newSingleThreadExecutor()（单个后台线程），它们均为大多数使用场景预定义了设置。否则，在手动配置和调整此类时，使用以下指导：
- 核心和最大池大小
- ThreadPoolExecutor 将根据 corePoolSize（参见
getCorePoolSize()）和 maximumPoolSize（参见getMaximumPoolSize()）设置的边界自动调整池大小。当新任务在方法execute(java.lang.Runnable)中提交时，如果运行的线程少于 corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程。如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。在大多数情况下，核心和最大池大小仅基于构造来设置，不过也可以使用setCorePoolSize(int)和setMaximumPoolSize(int)进行动态更改。 - 按需构造
- 默认情况下，即使核心线程最初只是在新任务到达时才创建和启动的，也可以使用方法
prestartCoreThread()或prestartAllCoreThreads()对其进行动态重写。如果构造带有非空队列的池，则可能希望预先启动线程。 - 创建新线程
- 使用
ThreadFactory创建新线程。如果没有另外说明，则在同一个ThreadGroup中一律使用Executors.defaultThreadFactory()创建线程，并且这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。通过提供不同的 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，等等。如果从 newThread 返回 null 时 ThreadFactory 未能创建线程，则执行程序将继续运行，但不能执行任何任务。 - 保持活动时间
- 如果池中当前有多于 corePoolSize 的线程，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止（参见
getKeepAliveTime(java.util.concurrent.TimeUnit)）。这提供了当池处于非活动状态时减少资源消耗的方法。如果池后来变得更为活动，则可以创建新的线程。也可以使用方法setKeepAliveTime(long, java.util.concurrent.TimeUnit)动态地更改此参数。使用 Long.MAX_VALUETimeUnit.NANOSECONDS的值在关闭前有效地从以前的终止状态禁用空闲线程。默认情况下，保持活动策略只在有多于 corePoolSizeThreads 的线程时应用。但是只要 keepAliveTime 值非 0，allowCoreThreadTimeOut(boolean)方法也可将此超时策略应用于核心线程。 - 排队
- 所有
BlockingQueue都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互：- 如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。
- 如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。
- 如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。
- 直接提交。工作队列的默认选项是
SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。 - 无界队列。使用无界队列（例如，不具有预定义容量的
LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。 - 有界队列。当使用有限的 maximumPoolSizes 时，有界队列（如
ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。
- 被拒绝的任务
当 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) 方法
- 此类提供 protected 可重写的
beforeExecute(java.lang.Thread, java.lang.Runnable)和afterExecute(java.lang.Runnable, java.lang.Throwable)方法，这两种方法分别在执行每个任务之前和之后调用。它们可用于操纵执行环境；例如，重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外，还可以重写方法terminated()来执行 Executor 完全终止后需要完成的所有特殊处理。
如果钩子 (hook) 或回调方法抛出异常，则内部辅助线程将依次失败并突然终止。
- 队列维护
- 方法
getQueue()允许出于监控和调试目的而访问工作队列。强烈反对出于其他任何目的而使用此方法。remove(java.lang.Runnable)和purge()这两种方法可用于在取消大量已排队任务时帮助进行存储回收。终止程序 AND 不再引用的池没有剩余线程会自动 shutdown。如果希望确保回收取消引用的池（即使用户忘记调用shutdown()），则必须安排未使用的线程最终终止：设置适当保持活动时间，使用 0 核心线程的下边界和/或设置allowCoreThreadTimeOut(boolean)。
方法：
protected void afterExecute(Runnable r, Throwable t) 基于完成执行给定 Runnable 所调用的方法。 void allowCoreThreadTimeOut(boolean value) 如果在保持活动时间内没有任务到达，新任务到达时正在替换（如果需要），则设置控制核心线程是超时还是终止的策略。 boolean allowsCoreThreadTimeOut() 如果此池允许核心线程超时和终止，如果在 keepAlive 时间内没有任务到达，新任务到达时正在替换（如果需要），则返回 true。 boolean awaitTermination(long timeout, TimeUnit unit) 请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行。 protected void beforeExecute(Thread t, Runnable r) 在执行给定线程中的给定 Runnable 之前调用的方法。 void execute(Runnable command) 在将来某个时间执行给定任务。 protected void finalize() 当不再引用此执行程序时，调用 shutdown。 int getActiveCount() 返回主动执行任务的近似线程数。 long getCompletedTaskCount() 返回已完成执行的近似任务总数。 int getCorePoolSize() 返回核心线程数。 long getKeepAliveTime(TimeUnit unit) 返回线程保持活动的时间，该时间就是超过核心池大小的线程可以在终止前保持空闲的时间值。 int getLargestPoolSize() 返回曾经同时位于池中的最大线程数。 int getMaximumPoolSize() 返回允许的最大线程数。 int getPoolSize() 返回池中的当前线程数。 BlockingQueue<Runnable> getQueue() 返回此执行程序使用的任务队列。 RejectedExecutionHandler getRejectedExecutionHandler() 返回用于未执行任务的当前处理程序。 long getTaskCount() 返回曾计划执行的近似任务总数。 ThreadFactory getThreadFactory() 返回用于创建新线程的线程工厂。 boolean isShutdown() 如果此执行程序已关闭，则返回 true。 boolean isTerminated() 如果关闭后所有任务都已完成，则返回 true。 boolean isTerminating() 如果此执行程序处于在 shutdown 或 shutdownNow 之后正在终止但尚未完全终止的过程中，则返回 true。 int prestartAllCoreThreads() 启动所有核心线程，使其处于等待工作的空闲状态。 boolean prestartCoreThread() 启动核心线程，使其处于等待工作的空闲状态。 void purge() 尝试从工作队列移除所有已取消的 Future 任务。 boolean remove(Runnable task) 从执行程序的内部队列中移除此任务（如果存在），从而如果尚未开始，则其不再运行。 void setCorePoolSize(int corePoolSize) 设置核心线程数。 void setKeepAliveTime(long time, TimeUnit unit) 设置线程在终止前可以保持空闲的时间限制。 void setMaximumPoolSize(int maximumPoolSize) 设置允许的最大线程数。 void setRejectedExecutionHandler(RejectedExecutionHandler handler) 设置用于未执行任务的新处理程序。 void setThreadFactory(ThreadFactory threadFactory) 设置用于创建新线程的线程工厂。 void shutdown() 按过去执行已提交任务的顺序发起一个有序的关闭，但是不接受新任务。 List<Runnable> shutdownNow() 尝试停止所有的活动执行任务、暂停等待任务的处理，并返回等待执行的任务列表。 protected void terminated() 当 Executor 已经终止时调用的方法。
第6部分 ScheduledExecutorService
ScheduledExecutorService是一个接口，它继承于于ExecutorService。它相当于提供了"延时"和"周期执行"功能的ExecutorService。
ScheduledExecutorService提供了相应的函数接口，可以安排任务在给定的延迟后执行，也可以让任务周期的执行。
ScheduledExecutorService函数列表
<V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) 创建并执行在给定延迟后启用的 ScheduledFuture。 ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) 创建并执行在给定延迟后启用的一次性操作。 ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) 创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行 ，接着在 initialDelay + 2 * period 后执行，依此类推。 ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) 创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。
第7部分 ScheduledThreadPoolExecutor
ThreadPoolExecutor，它可另行安排在给定的延迟后运行命令，或者定期执行命令。需要多个辅助线程时，或者要求 ThreadPoolExecutor 具有额外的灵活性或功能时，此类要优于 Timer。
一旦启用已延迟的任务就执行它，但是有关何时启用，启用后何时执行则没有任何实时保证。按照提交的先进先出 (FIFO) 顺序来启用那些被安排在同一执行时间的任务。
虽然此类继承自 ThreadPoolExecutor，但是几个继承的调整方法对此类并无作用。特别是，因为它作为一个使用 corePoolSize 线程和一个无界队列的固定大小的池，所以调整 maximumPoolSize 没有什么效果。
ScheduledThreadPoolExecutor函数列表
构造方法：
ScheduledThreadPoolExecutor(int corePoolSize) 使用给定核心池大小创建一个新 ScheduledThreadPoolExecutor。 ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) 使用给定初始参数创建一个新 ScheduledThreadPoolExecutor。 ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) 使用给定的初始参数创建一个新 ScheduledThreadPoolExecutor。 ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler) 使用给定初始参数创建一个新 ScheduledThreadPoolExecutor。
方法：
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) 修改或替换用于执行 callable 的任务。 protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) 修改或替换用于执行 runnable 的任务。 void execute(Runnable command) 使用所要求的零延迟执行命令。 boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() 获取有关在此执行程序已 shutdown 的情况下、是否继续执行现有定期任务的策略。 boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() 获取有关在此执行程序已 shutdown 的情况下是否继续执行现有延迟任务的策略。 BlockingQueue<Runnable> getQueue() 返回此执行程序使用的任务队列。 boolean remove(Runnable task) 从执行程序的内部队列中移除此任务（如果存在），从而如果尚未开始，则其不再运行。 <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) 创建并执行在给定延迟后启用的 ScheduledFuture。 ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) 创建并执行在给定延迟后启用的一次性操作。 ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) 创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行 ，接着在 initialDelay + 2 * period 后执行，依此类推。 ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) 创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。 void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) 设置有关在此执行程序已 shutdown 的情况下是否继续执行现有定期任务的策略。 void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) 设置有关在此执行程序已 shutdown 的情况下是否继续执行现有延迟任务的策略。 void shutdown() 在以前已提交任务的执行中发起一个有序的关闭，但是不接受新任务。 List<Runnable> shutdownNow() 尝试停止所有正在执行的任务、暂停等待任务的处理，并返回等待执行的任务列表。 <T> Future<T> submit(Callable<T> task) 提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future。 Future<?> submit(Runnable task) 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。 <T> Future<T> submit(Runnable task, T result) 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
第8部分 Executors
Executors是个静态工厂类。它通过静态工厂方法返回Executor、ExecutorService、ScheduledExecutorService、ThreadFactory 和 Callable等类的对象。
Executors函数列表
static Callable<Object> callable(PrivilegedAction<?> action) 返回 Callable 对象，调用它时可运行给定特权的操作并返回其结果。 static Callable<Object> callable(PrivilegedExceptionAction<?> action) 返回 Callable 对象，调用它时可运行给定特权的异常操作并返回其结果。 static Callable<Object> callable(Runnable task) 返回 Callable 对象，调用它时可运行给定的任务并返回 null。 static <T> Callable<T> callable(Runnable task, T result) 返回 Callable 对象，调用它时可运行给定的任务并返回给定的结果。 static ThreadFactory defaultThreadFactory() 返回用于创建新线程的默认线程工厂。 static ExecutorService newCachedThreadPool() 创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。 static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) 创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们，并在需要时使用提供的 ThreadFactory 创建新线程。 static ExecutorService newFixedThreadPool(int nThreads) 创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程。 static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) 创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程，在需要时使用提供的 ThreadFactory 创建新线程。 static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) 创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。 static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) 创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。 static ExecutorService newSingleThreadExecutor() 创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。 static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) 创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程，并在需要时使用提供的 ThreadFactory 创建新线程。 static ScheduledExecutorService newSingleThreadScheduledExecutor() 创建一个单线程执行程序，它可安排在给定延迟后运行命令或者定期地执行。 static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) 创建一个单线程执行程序，它可安排在给定延迟后运行命令或者定期地执行。 static <T> Callable<T> privilegedCallable(Callable<T> callable) 返回 Callable 对象，调用它时可在当前的访问控制上下文中执行给定的 callable 对象。 static <T> Callable<T> privilegedCallableUsingCurrentClassLoader(Callable<T> callable) 返回 Callable 对象，调用它时可在当前的访问控制上下文中，使用当前上下文类加载器作为上下文类加载器来执行给定的 callable 对象。 static ThreadFactory privilegedThreadFactory() 返回用于创建新线程的线程工厂，这些新线程与当前线程具有相同的权限。 static ExecutorService unconfigurableExecutorService(ExecutorService executor) 返回一个将所有已定义的 ExecutorService 方法委托给指定执行程序的对象，但是使用强制转换可能无法访问其他方法。 static ScheduledExecutorService unconfigurableScheduledExecutorService(ScheduledExecutorService executor) 返回一个将所有已定义的 ExecutorService 方法委托给指定执行程序的对象，但是使用强制转换可能无法访问其他方法。
参考
1.jdk api