(九) 执行器(线程池)
线程池：
1.使用线程池(Thread Pool)的前提是，程序中创建了大量的生命周期很短的线程。构建一个新的线程是有一定代价的，因为涉及与操作系统的交互。
2.另一个使用线程池的理由是减少并发线程的数目。创建大量线程会大大降低性能甚至使虚拟机崩溃。所以，当有一个会创建许多线程的算法，应该使用一个线程数“固定的”线程池以限制并发总数。
3.线程池中包含许多准备运行的空闲线程，将Runnable对象交给线程池，就会有一个线程调用run方法。当run方法退出时，线程不会死亡，而是在池中准备为下一个请求服务。
执行器(Executor)类中有许多静态工厂方法用来创建线程池
方法 描述
newCachedThreadPool 必要时创建新线程；空闲线程会被保留60秒。
newFixedThreadPool 该池包含固定数量的线程；空闲线程会一直被保留。
newSingleThreadPool 只有一个线程的“池”，该线程顺序执行每一个提交的任务
newScheduledThreadPool 用于预定执行而构建的固定线程池，替代java.util.Timer
newSingleThreadScheduledExecutor 用于预定执行而构建的单线程“池”
1.线程池
(1)newCachedThreadPool方法构建了一个线程池，对于每个任务，如果有空闲线程可用，立即让它执行任务，如果没有可用的空闲线程，则创建一个新线程。
(2)newFixedThreadPool方法构建一个具有固定大小的线程池。如果提交的任务数多余空闲的线程数，那么把得不到服务的任务放置到队列中，当其他任务完成后在运行它。
(3)newSingleThreadPool方式是一个退化了的大小为1的线程池。由一个线程执行提交的任务，一个接着一个。
以上三个方法返回实现了ExecutorService接口的ThreadPoolExecutor类的对象。
3种将Runnable对象或Callable对象提交给ExecutorService的方法
(1)Future<?> submit(Runnable task)
该方法返回Future<?>，该对象可以调用isDone,cancel或isCancelled方法，但是当get方法完成时，只是简单的返回null。
(2)Future<T> submit(Runnable task, T result)
该方法提交Runnable对象，并且Future的get方法完成时返回指定的result对象。
(3)Future<T> submit(Callable<T> task)
该方法提交Callable对象，并且Future的get方法将在计算结果准备好的时候得到结果。
shutdown : 当用完一个线程池的时候，需要调用shutdown方法。该方法启动该池的关闭序列。被关闭的执行器不再接受新的任务。当所有任务都完成之后，线程池中的线程死亡。
shutdownNow : 该池取消尚未开始的所有任务并试图中断正在运行的线程。
连接线程池的步骤：
(1)调用Executors类中的静态方法newCachedThreadPool或newFixedThreadPool。
(2)调用submit提交Runnable或Callable对象。
(3)如果想要取消一个任务，或如果提交Callable对象，那就要保存好返回的Future对象。
(4)当不再提交任何任务时，调用shutdown。
import java.io.File;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
public class FutureTest {
public static void main(String[] args) {
String dirctory = "";
String keyword = null;
File dir = new File(dirctory);
Scanner in = new Scanner(System.in);
while(!dir.exists() || !dir.isDirectory()){
System.out.println("Enter base directory (e.g. D:\\WorkSpace\\Demo):");
dirctory = in.nextLine();
dir = new File(dirctory);
}
while(keyword==null || "".equals(keyword)){
System.out.println("Enter keyword(e.g. python):");
keyword = in.nextLine();
}
System.out.println("The directory: "+dir.getPath()+", The keyword: "+keyword);
//1.使用FutureTask方式
// MatchCounter counter = new MatchCounter(new File(dirctory), keyword);
// FutureTask<Integer> task = new FutureTask<Integer>(counter);
// Thread t = new Thread(task);
// t.start();
//2.使用线程池
ExecutorService pool = Executors.newCachedThreadPool();
MatchCounter counter = new MatchCounter(dir, keyword, pool);
Future<Integer> result = pool.submit(counter);
try{
System.out.println(result.get() + " matching files.");
}catch(Exception e){
e.printStackTrace();
}
//2.使用线程池应关闭线程池
pool.shutdown();
int largestPoolSize = ((ThreadPoolExecutor)pool).getLargestPoolSize();
System.out.println("largest pool size: " + largestPoolSize);
}
}
public class MatchCounter implements Callable<Integer>{
private ExecutorService pool;
private File directory;
private String keyword;
private int count;
public MatchCounter(File directory, String keyword){
this.directory = directory;
this.keyword = keyword;
count = 0;
}
public MatchCounter(File directory, String keyword, ExecutorService pool){
this.directory = directory;
this.keyword = keyword;
this.pool = pool;
count = 0;
}
@Override
public Integer call() throws Exception {
try{
File[] files = directory.listFiles();
List<Future<Integer>> results = new ArrayList<Future<Integer>>();
for(File file:files){
if(file.isDirectory()){
//1.通过FutureTask
// MatchCounter counter = new MatchCounter(file, keyword);
// FutureTask task = new FutureTask(counter);
// results.add(task);
// Thread t = new Thread(task);
// t.start();
//2.通过线程池
MatchCounter counter = new MatchCounter(file, keyword, pool);
results.add(pool.submit(counter));
}else{
if(search(file)){
count++;
}
}
}
for(Future<Integer> result:results){
try{
count += result.get();
}catch(Exception e){
e.printStackTrace();
}
}
}catch(Exception e){
e.printStackTrace();
}
return count;
}
private boolean search(File file) {
try{
Scanner in = new Scanner(new FileInputStream(file));
boolean found = false;
while(!found && in.hasNextLine()){
String line = in.nextLine();
if(line.contains(keyword)){
found = true;
}
}
in.close();
return found;
}catch(Exception e){
return false;
}
}
}
2.预订执行
ScheduledExecutorService接口具有为预定执行(Scheduled Execution)或重复执行任务而设计的方法。它是一种允许使用线程池机制的java.util.Timer的泛化。
Executor类的newScheduledThreadPool和newSingleThreadScheduledExecutor方法实现了ScheduledExecutorService接口的对象。
可以预订Runnable或Callable在初始的延迟之后之后只运行一次。也可以预定一个Runnable对象周期性的运行。
3.控制任务组
使用执行器有更有实际意义的原因
(1)可以在执行器中使用shutdownNow方法取消所有的任务。
(2)invokeAny方法提交所有对象到一个Callable对象的集合中，并返回某个已经完成了的任务的结果。
(3)invokeAll方法提交所有的对象到一个Callable对象的集合中，并返回一个Future对象的列表，代表所有任务的解决方案。
e.g. List<Callable<T>> task = ...;
List<Future<T>> results = executor.invokeAll(tasks);
for(Future<T> result : results){
processFuture(result.get());
}
这个方法的缺点是，如果第一个任务占用了很多时间执行，则可能不得不等待。可以用ExecutorCompletionService来进行排列。
(4)ExecutorCompletionService的构建方法
e.g. ExecutorCompletionService service = new ExecutorCompletionService(executor);
for(Callable<T> task:tasks){
service.submit(task);
}
for(int i=0;i<tasks.size();i++){
processFuture(service.task().get());
}
JDK中的表述十分清晰
一个 ExecutorService，它使用可能的几个池线程之一执行每个提交的任务，通常使用 Executors 工厂方法配置。
线程池可以解决两个不同问题：由于减少了每个任务调用的开销，它们通常可以在执行大量异步任务时提供增强的性能，并且还可以提供绑定和管理资源（包括执行任务集时使用的线程）的方法。每个 ThreadPoolExecutor 还维护着一些基本的统计数据，如完成的任务数。
为了便于跨大量上下文使用，此类提供了很多可调整的参数和扩展钩子 (hook)。但是，强烈建议程序员使用较为方便的 Executors 工厂方法 Executors.newCachedThreadPool()（*线程池，可以进行自动线程回收）、Executors.newFixedThreadPool(int)（固定大小线程池）和 Executors.newSingleThreadExecutor()（单个后台线程），它们均为大多数使用场景预定义了设置。否则，在手动配置和调整此类时，使用以下指导：
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 未能创建线程，则执行程序将继续运行，但不能执行任何任务。getKeepAliveTime(java.util.concurrent.TimeUnit)）。这提供了当池处于非活动状态时减少资源消耗的方法。如果池后来变得更为活动，则可以创建新的线程。也可以使用方法 setKeepAliveTime(long, java.util.concurrent.TimeUnit) 动态地更改此参数。使用 Long.MAX_VALUE TimeUnit.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 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。
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 类也是可能的，但这样做需要非常小心，尤其是当策略仅用于特定容量或排队策略时。beforeExecute(java.lang.Thread, java.lang.Runnable) 和 afterExecute(java.lang.Runnable, java.lang.Throwable) 方法，这两种方法分别在执行每个任务之前和之后调用。它们可用于操纵执行环境；例如，重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外，还可以重写方法 terminated() 来执行 Executor 完全终止后需要完成的所有特殊处理。
如果钩子 (hook) 或回调方法抛出异常，则内部辅助线程将依次失败并突然终止。
getQueue() 允许出于监控和调试目的而访问工作队列。强烈反对出于其他任何目的而使用此方法。remove(java.lang.Runnable) 和 purge() 这两种方法可用于在取消大量已排队任务时帮助进行存储回收。shutdown()），则必须安排未使用的线程最终终止：设置适当保持活动时间，使用 0 核心线程的下边界和/或设置 allowCoreThreadTimeOut(boolean)。 上一篇： (五) 同步