Java 线程（2）- Executor 框架

当我们需要一定数量的线程来处理问题的时候，我们需要考虑：

操作系统对线程的数量的限制
创建、调度和终止线程的系统开销
线程本身对系统资源的消耗（尤其是内存，JVM 需要为每个线程维护一个独立的线程栈 -Xss<size>）

利用线程池（Thread Pool）来管理多个线程是一个不错的选择，而 Java 的 Executor Framework 为我们提供了许多有用的方法来帮助我们建立和使用线程池

Executor Framework

executor

Executor Framework 中的接口和类都定义在 java.util.concurrent 包中，其基本接口包括：Executor、ThreadFactory 和 Callable，其中 Executor 接口代表了线程池，ThreadFactory 用于创建线程，Callable 接口代表一个可执行的任务（task），其 call 方法用于执行任务，可以返回结果和抛出异常。ExecutorService 接口继承自 Executor 并添加了关闭（shutdown）线程池、提交（submit）任务、调用（invoke）任务和等待任务结束的方法（await termination），而 ScheduledExecutorService 接口在 ExecutorService 接口的基础之上有提供了按时间调度（schedule）任务的方法。而 Executors 类提供了各种工厂方法来实列化这些接口。

有些时候，Callable 接口比 java.lang.Runable 接口更实用，因为 java.lang.Runable 中的 run 方法既不能返回结果也不能抛出异常，这就有可能要求我们使用共享变量来在多线程之家共享结果，或需要在 run 方法中记录异常信息

Executors

Executors 类基于 ThreadPoolExecutor 提供了许多开箱即用的线程池

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

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

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));

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

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}

public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1, threadFactory));
}

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

使用 ScheduledThreadPoolExecutor 要比用 java.util.Timer 更好，因为 Timer 使用一个线程执行定时任务，并且 Timer 可能抛出 unexpected exception

配置 ThreadPoolExecutor

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

corePoolSize 和 maximumPoolSize，其中 corePoolSize 是线程池的目标大小，即使线程池中没有任务执行，线程池也会维护和 corePoolSize 相同的线程（当线程池创建时, 其默认不会立即创建出和 corePoolSize 相同的线程, 除非调用 prestartAllCoreThreads 方法），而 maximumPoolSize 指定了线程池中同时可执行的线程的上限。当在线程池中提交一个任务请求，如果此时的执行线程数小于 corePoolSize，线程会为该任务创建一个新的线程（即使执行线程中有空闲的线程），如果此时执行线程数大于 corePoolSize，并且小于 maximumPoolSize，这时线程池会根据 workQueue 的情况来判断是否需要创建新线程：如果新的请求可以被排队，线程池优先将新的请求做排队处理；如果新的请求无法被排队（如: 队列已经满了），如果此时执行线程数小于 maximumPoolSize，则线程池为该请求创建新的线程，否则的话，线程池会拒绝该请求，交由相应的 RejectedExecutionHandler 处理
keepAliveTime，如果线程池中的线程的空闲时间超过 keepAliveTime，线程池会将其作为回收的候选线程，例如：如果线程池中的线程数超过了 corePoolSize，该线程就会被回收
workQueue，使用队列来保存（排队）任务，明显要优于直接把任务放在线程中，对于workQueue 的选择通常有三种策略：
1. Direct handoffs，该策略适用于相互有一定依赖的任务，workQueue 一般使用 SynchronousQueue，并且线程池的大小一般是没有限制的（corePoolSize=0 并且 maximumPoolSize=Integer.MAX_VALUE)，通常还需要配合 keepAliveTime，放入池中的任务会被立即分配线程执行
2. Unbounded queues，该策略适用于相互独立的线程，如果池中的线程数达到阀值，新进的线程会被放入队列中排队
3. Bounded queues，该策略有助于帮助避免系统资源耗尽，如把池的大小设置的很小，但 queue 却很大，这时可以有效地利用系统资源，但代价是降低了系统的吞吐量
threadFactory，用来创建线程池中的线程，默认通过调用 defaultThreadFactory() 方法获得，可以通过实现 ThreadFactory 接口自定义 Thread Factory
Rejected handler (饱和策略) 包括
1. ThreadPoolExecutor.AbortPolicy，默认策略，如果线程池饱和，则抛出 RejectedExecutionException
2. ThreadPoolExecutor.DiscardPolicy，如果线程池饱和，线程池会丢弃新的任务请求
3. ThreadPoolExecutor.DiscardOldestPolicy，如果线程池饱和，线程池会优先丢弃任务队里中的头请求
4. ThreadPoolExecutor.CallerRunsPolicy，该策略可以用来调节任务的提交速率，类似于断路器，如果线程池饱和，线程池会将新增的任务请求推回给调用者，从而减缓客户端的任务请求

线程池最佳实践：当线程池中的任务大小均匀（homogeneous）并且没有相互依赖时线程池可以获得最佳的性能

确定线程池大小

如果线程池太大（too big）可能会耗尽系统资源，如果太小（too small），则会影响程序的吞吐量，可以从线程池中的任务类型来考虑线程池的大小：

如果是计算密集型（compute-intensive）任务设置为：Runtime.getRuntime().availableProcessors() + 1
如果是 I/O 密集型或是阻塞型任务，则可能需要一个大的线程池，可以通过以下方式来估计线程池大小

$N_{threads} = N_{cpu} \times U_{cpu} \times (1 + \frac{W}{C})$

$N_{cpu} = \text{number of CPUs}$

$U_{cpu} = \text{target CPU utilization, } 0 \le U_{cpu} \le 1$

$\frac{W}{C} =\text{ratio of wait time to compute time}$

CPUs 数量可以从 Runtime.getRuntime().availableProcessors() 获得

关闭线程池

我们可以通过 ExcutorService.shutdown() 或 ExcutorService.shutdownNow() 来关闭线程池，当调用 shutdown() 方法关闭线程池后，线程池只是不接受任何新任务，该方法不会等待那些已经开始但尚未结束的任务。所以如果在关闭线程池时，需要结束所有已经提交的任务（包括结束的、还在运行的和正在等待执行的），则需要需要记录所有已经提交的任务，然后取消那些尚未结束的任务，比如：

for (SomeTask task : submittedTasks) {
            if (!task.isFinished()) {
                task.cancelTask();
            }
 }
service.shutdown();

如果需要等待那些已经开始的任务完成，则需要调用 awaitTermination，比如：

ExecutorService exec = Executors.newCachedThreadPool(); 
try {
    for (final Runnable task : tasks) {
        exec.execute(task); 
    }
} finally { 
    exec.shutdown(); // shutdown the service
    exec.awaitTermination(timeout, unit); 
}

当调用 shutdownNow() 时，其会返回一个没有执行的（等待）任务列表，同 shutdown() 一样，其不会等待那些已经开始执行，但尚未完成的任务，所以没有办法获得完成任务列表，因此仍需要和 shutdown() 一样去处理那些已经开始执行，但尚未完成的任务。

Java 线程（2）- Executor 框架

Java 线程（2）- Executor 框架

Executor Framework

Executors

配置 ThreadPoolExecutor

确定线程池大小

关闭线程池

参考

推荐阅读更多精彩内容