Java的多线程之执行器

构建一个新的线程是有一定代价的，因为涉及与操作系统的交互。如果程序中创建了大量的生命期很短的线程，应该使用线程池(thread pool)。一个线程池中包含许多准备运行的空闲线程。将Runnable对象交给线程池，就会有一个线程调用run方法。当run方法退出时，线程不会死亡，而是在池中准备为下一个请求提供服务。

另一个使用线程池的理由是减少并发线程的数目。创建大量线程会大大降低性能甚至使虚拟机崩溃。如果有一个会创建许多线程的算法，应该使用一个线程数“固定的”线程池以限制并发线程的总数。

执行器(Executor)类有许多静态工厂方法用来构建线程池。如下：



一、线程池

newCachedThreadPool方法构建一个线程池，对于每个任务，如果有空闲线程可用，立即让它执行任务，如果没有可用的空闲线程，则创建一个新线程。

newFixedThreadPool方法构建一个具有固定大小的线程池。如果提交的任务数多于空闲的线程数，那么把得不到服务的任务放置到队列中。当其他任务完成以后再运行它们。

newSingleThreadExecutor是一个退化了的大小为1的线程池；由一个线程执行提交的任务，一个接着一个。这3个方法返回实现了ExecutorService接口的ThreadPoolExecutor类的对象。

可用下面的方法之一将一个Runnable对象或Callable对象提交给ExecutorService：

Future<?> submit(Runnable task)

Future<T> submit(Runnable task,T result)

Future<T> submit(Callable<T> task)

该池会在方便的时候尽早执行提交的任务。调用submit时，会得到一个Future对象，可用来查询该任务的状态。

第一个submit方法返回对象可用用来调用isDone、cancel或isCancelled。但是，get方法在完成时只是简单的返回null。

第二个方法的返回值Future的get方法在完成的时候返回指定的result对象。

第三个方法返回值Future对象将在计算结果准备好的时候得到它。

当用完一个线程池的时候，调用shutdown。该方法启动该池的关闭序列。被关闭的执行器不在接受新的任务。当所有任务完成以后，线程池中的线程死亡。另一种方法是调用shutdownNow。该池取消尚未开始的所有任务并试图中断正在运行的线程。

下面总结了在使用线程池时应该做的事：

1>调用Executors类中静态的方法newCachedThreadPool或newFixedThreadPool。

2>调用submit提交Runnable或Callable对象。

3>如果想要取消一个任务，或如果提交Callable对象，那就要保持好返回的Future对象。

4>当不再提交任何任务时，调用shutdown。

package com.thread.threadPool;

import java.io.File;
import java.util.Scanner;
import java.util.concurrent.ExecutionException;
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) {
Scanner in = new Scanner(System.in);
System.out.println("请输入目录:");
String directory = in.nextLine();
System.out.println("请输入关键字:");
String keyword = in.nextLine();
in.close();
ExecutorService pool = Executors.newCachedThreadPool();

MatchCounter counter = new MatchCounter(new File(directory), keyword,
pool);
Future<Integer> result = pool.submit(counter);
try {
System.out.println(result.get() + " matching files.");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
pool.shutdown();
int largestPoolSize = ((ThreadPoolExecutor)pool).getLargestPoolSize();
System.out.println("largest pool size="+largestPoolSize);
}
}

package com.thread.threadPool;

import java.io.File;
import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

public class MatchCounter implements Callable<Integer> {

private File directory;
private String keyword;
private ExecutorService pool;
private int count;

public MatchCounter(File directory, String keyword, ExecutorService pool) {
this.directory = directory;
this.keyword = keyword;
this.pool = pool;
}

@Override
public Integer call() {
count = 0;
try {
File[] files = directory.listFiles();
List<Future<Integer>> results = new ArrayList<>();

for (File file : files) {
if (file.isDirectory()) {
MatchCounter counter = new MatchCounter(file, keyword, pool);
Future<Integer> result = pool.submit(counter);
results.add(result);
} else {
if (search(file))
count++;
}
}
for (Future<Integer> result : results) {
try {
count += result.get();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
} catch (InterruptedException e) {
}
return count;

}

public boolean search(File file) {
try (Scanner in = new Scanner(file)) {
boolean found = false;
while (!found && in.hasNext()) {
String line = in.nextLine();
if (line.contains(keyword))
return found = true;
}
return found;
} catch (FileNotFoundException e) {
e.printStackTrace();
return false;
}
}
}

二、预定执行

ScheduledExecutorService接口具有为预定执行（Scheduled Execution）或重复执行任务而设计的方法。它是一种允许使用线程池机制的java.util.Timer的泛化。

Executors类的newScheduledThreadPool和newSingleThreadScheduledExecutor方法将返回实现了ScheduledExecutorService接口的对象。

可以预定Runnable或Callable在初始的延迟之后只运行一次。也可以预定一个Runnable对象周期性地运行。

三、控制任务组

一个执行器服务作为线程池使用，以提高执行任务的效率。有时候，使用执行器有更有实际意义的原因，控制一组相关任务。

invokeAny方法提交所有对象到一个Callable对象的集合中，并返回某个已经完成了的任务的结果。无法知道返回的究竟是哪个任务的结果。

invokeAll方法提交所有对象到一个Callable对象的集合中，并返回一个Future对象的列表，代表所有任务的解决方案。当计算结果可获得时，可以如下处理结果：

List<Callable<T>> tasks = ...;
List<Future<T>> results = executor.invokeAll(tasks);
for (Future<T> result : results) {
processFurther(result.get());
}

这个方法的缺点是如果第一个任务恰巧花去了很多时间，则可能不得不进行等待。将结果按可获得的顺序保存起来更有实际意义。可以用ExecutorCompletionService来进行排序。

用常规的方法获得一个执行器。然后，构建一个ExecutorCompletionService，提交任务给完成服务（completion service）。该服务管理Future对象的阻塞队列，其中包含已经提交的任务的执行结果。更有效的组织形式如下：

ExecutorCompletionService service = new ExecutorCompletionService(executor);
for (Callable<T> task : tasks)
service.submit(task);
for (int i = 0; i < tasks.size(); i++)
processFurther(service.take().get());

四、fork-join框架