在之前的例子中，我们使用执行器框架都是在主类中提交任务，等待任务执行完毕后再去处理任务执行的结果。接下来我们打算将任务的提交和结果的处理都放置到线程中去执行。在每个任务内部提交自己到执行器，然后通过一个统一的结果处理线程来处理所有任务执行的结果。
为了解决这个问题，执行器框架为我们提供了一个CompletionService类，任务执行线程和结果处理线程能够共享这个类，结果处理线程便可以在这里渠道已经执行完毕的任务的结果。CompletionService类的内部也是通过一个ExecutorService来提交任务的。
首先，创建任务线程，实现Callable接口。模拟报表生成过程。
/**
* 模拟生成报告
*
* Created by hadoop on 2016/11/3.
*/
public class ReportGenerator implements Callable<String> {
private String sender;
private String title;
public ReportGenerator(String sender, String title) {
this.sender = sender;
this.title = title;
}
@Override
public String call() throws Exception {
long duration = (long)(Math.random() * 10);
System.out.printf("ReportGenerator: Generator report %s_%s duration %d seconds.\n", sender, title, duration);
TimeUnit.SECONDS.sleep(duration);
return sender + "_" + title;
}
}
然后我们创建任务提交线程，这个线程的构造方法接受两个参数，分别是报表名称和CompletionService对象。将报表生成任务提交到CompletionService去执行。
import java.util.concurrent.CompletionService;
/**
* 模拟请求获取报告
*
* Created by hadoop on 2016/11/3.
*/
public class ReportRequest implements Runnable {
private String name;
private CompletionService<String> service;
public ReportRequest(String name, CompletionService<String> service) {
this.name = name;
this.service = service;
}
@Override
public void run() {
ReportGenerator generator = new ReportGenerator(name, "Report");
service.submit(generator);
}
}
下面我们创建任务结果处理类，来打印生成的报表。这个类同样会拿到CompletionService的引用，然后循环调用CompletionService.poll()方法来从任务结果队列中获取执行的结果，这个方法接受一个时间参数，如果当前结果队列为空，那么则等待这个时间，超时返回null。不带参数的poll()方法，如果对别为空则直接返回null。
/**
* 处理报表结果
*
* Created by hadoop on 2016/11/3.
*/
public class ReportProcessor implements Runnable {
private boolean end;
private CompletionService<String> service;
public ReportProcessor(boolean end, CompletionService<String> service) {
this.end = end;
this.service = service;
}
@Override
public void run() {
while (!end) {
try {
Future<String> future = service.poll(20, TimeUnit.SECONDS);
if (future != null) {
System.out.printf("ReportReceiver: received %s\n", future.get());
}
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
public void setEnd(boolean end) {
this.end = end;
}
}
最后我们创建主方法类。在这里我们创建ExecutorServer并把它赋值给ExecutorCompletionService。之后创建两个报表请求任务和一个报表处理任务，同时持有ExecutorCompletionService的引用。
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* 在执行器中分离任务执行和结果处理
*
* 我们如何处理在一个对象里发送任务给执行器，在另一个对象里处理任务执行结果。
* 对于这种情况Java API提供了CompletionService类
*
* CompletionService使用Executor对象类执行任务。
* 优势在于：可以共享CompletionService。
* 缺点在于：CompletionService获取的Future对象只能是已经执行完毕的任务，他没有办法控制任务状态，只能处理任务结果。
*
* 我们创建了一个ExecutorService，然后使用这个ExecutorService来初始化一个ExecutorCompletionService<String>(executor)。
*
* 首先创建了两个ReportRequest任务，然后在任务内部使用service.submit(generator)方法调用报表生成任务。
*
* 然后再ReportProcessor中不断调用service.poll(20, TimeUnit.SECONDS);方法获取已经执行完的结果，如果当前没有结果那么等待20秒。
*
* CompletionService还提供了两个人方法：
* poll()：如果没有任何Future直接返回null。
* take()：如若任务队列中没有Future那么阻塞知道有可用的Future。
*
* Created by hadoop on 2016/11/3.
*/
public class Main {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
CompletionService<String> service = new ExecutorCompletionService<String>(executor);
ReportRequest request1 = new ReportRequest("Face", service);
ReportRequest request2 = new ReportRequest("Online", service);
ReportProcessor processor = new ReportProcessor(false, service);
Thread thread1 = new Thread(request1);
Thread thread2 = new Thread(request2);
Thread thread3 = new Thread(processor);
thread1.start();
thread2.start();
thread3.start();
try {
thread1.join();
thread2.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
executor.shutdown();
try {
executor.awaitTermination(1, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
processor.setEnd(true);
}
}
控制台中，我们可以看到两个任务的提交信息和结果处理信息。
ReportGenerator: Generator report Online_Report duration 4 seconds.
ReportGenerator: Generator report Face_Report duration 1 seconds.
ReportReceiver: received Face_Report
ReportReceiver: received Online_Report
