线程执行者（十一）执行者分离任务的启动和结果的处理
执行者分离任务的启动和结果的处理
通常，当你使用执行者执行并发任务时，你将会提交 Runnable或Callable任务给这个执行者，并获取Future对象控制这个方法。你可以发现这种情况，你需要提交任务给执行者在一个对象中，而处理结果在另一个对象中。基于这种情况，Java并发API提供CompletionService类。
CompletionService 类有一个方法来提交任务给执行者和另一个方法来获取已完成执行的下个任务的Future对象。在内部实现中，它使用Executor对象执行任务。这种行为的优点是共享一个CompletionService对象，并提交任务给执行者，这样其他（对象）可以处理结果。其局限性是，第二个对象只能获取那些已经完成它们的执行的任务的Future对象，所以，这些Future对象只能获取任务的结果。
在这个指南中，你将学习如何使用CompletionService类把执行者启动任务和处理它们的结果分开。
准备工作…
这个指南的例子使用Eclipse IDE实现。如果你使用Eclipse或其他IDE，如NetBeans，打开它并创建一个新的Java项目。
如何做…
按以下步骤来实现的这个例子:
1.创建ReportGenerator类，并指定其实现Callable接口，参数化为String类型。
1
|
public class ReportGenerator implements Callable<String> {
|
2.声明两个私有的、String类型的属性，sender和title，用来表示报告的数据。
1
|
private String sender;
|
2
|
private String title;
|
3.实现这个类的构造器，初始化这两个属性。
1
|
public ReportGenerator(String sender, String title){
|
2
|
this.sender=sender;
|
3
|
this.title=title;
|
4
|
}
|
4.实现call()方法。首先，让线程睡眠一段随机时间。
1
|
@Override
|
2
|
public String call() throws Exception {
|
3
|
try {
|
4
|
Long duration=(long)(Math.random()*10);
|
5
|
System.out.printf("%s_%s: ReportGenerator: Generating a report during %d seconds\n",this.sender,this.title,duration);
|
6
|
TimeUnit.SECONDS.sleep(duration);
|
7
|
} catch (InterruptedException e) {
|
8
|
e.printStackTrace();
|
9
|
}
|
5.然后，生成一个有sender和title属性的字符串的报告，返回这个字符串。
1
|
String ret=sender+": "+title;
|
2
|
return ret;
|
3
|
}
|
6.创建ReportRequest类，实现Runnable接口。这个类将模拟一些报告请求。
1
|
public class ReportRequest implements Runnable {
|
7.声明私有的、String类型的属性name，用来存储ReportRequest的名称。
1
|
private String name;
|
8.声明私有的、CompletionService类型的属性service。CompletionService接口是个参数化接口，使用String类型参数化它。
1
|
private CompletionService<String> service;
|
9.实现这个类的构造器，初始化这两个属性。
1
|
public ReportRequest(String name, CompletionService<String> service){
|
2
|
this.name=name;
|
3
|
this.service=service;
|
4
|
}
|
10.实现run()方法。创建1个ReportGenerator对象，并使用submit()方法把它提交给CompletionService对象。
1
|
@Override
|
2
|
public void run() {
|
3
|
ReportGenerator reportGenerator=new ReportGenerator(name,"Report");
|
4
|
service.submit(reportGenerator);
|
5
|
}
|
11.创建ReportProcessor类。这个类将获取ReportGenerator任务的结果，指定它实现Runnable接口。
1
|
public class ReportProcessor implements Runnable {
|
12.声明一个私有的、CompletionService类型的属性service。由于CompletionService接口是个参数化接口，使用String类作为这个CompletionService接口的参数。
1
|
private CompletionService<String> service;
|
13.声明一个私有的、boolean类型的属性end。
1
|
private boolean end;
|
14.实现这个类的构造器，初始化这两个属性。
1
|
public ReportProcessor (CompletionService<String> service){
|
2
|
this.service=service;
|
3
|
end=false;
|
4
|
}
|
15.实现run()方法。当属性end值为false，调用CompletionService接口的poll()方法，获取CompletionService执行的下个已完成任务的Future对象。
1
|
@Override
|
2
|
public void run() {
|
3
|
while (!end){
|
4
|
try {
|
5
|
Future<String> result=service.poll(20, TimeUnit.SECONDS);
|
16.然后，使用Future对象的get()方法获取任务的结果，并且将这些结果写入到控制台。
01
|
if (result!=null) {
|
02
|
String report=result.get();
|
03
|
System.out.printf("ReportReceiver: Report Received:%s\n",report);
|
04
|
}
|
05
|
} catch (InterruptedException | ExecutionException e) {
|
06
|
e.printStackTrace();
|
07
|
}
|
08
|
}
|
09
|
System.out.printf("ReportSender: End\n");
|
10
|
}
|
17.实现setEnd()方法，用来修改属性end的值。
1
|
public void setEnd(boolean end) {
|
2
|
this.end = end;
|
3
|
}
|
18.实现这个示例的主类，通过创建Main类，并实现main()方法。
1
|
public class Main {
|
2
|
public static void main(String[] args) {
|
19.使用Executors类的newCachedThreadPool()方法创建ThreadPoolExecutor。
1
|
ExecutorService executor=(ExecutorService)Executors.newCachedThreadPool();
|
20.创建CompletionService，使用前面创建的执行者作为构造器的参数。
1
|
CompletionService<String> service=new ExecutorCompletionService<>(executor);
|
21.创建两个ReportRequest对象，并用线程执行它们。
1
|
ReportRequest faceRequest=new ReportRequest("Face", service);
|
2
|
ReportRequest onlineRequest=new ReportRequest("Online";,service);
|
3
|
Thread faceThread=new Thread(faceRequest);
|
4
|
Thread onlineThread=new Thread(onlineRequest);
|
22.创建一个ReportProcessor对象，并用线程执行它。
1
|
ReportProcessor processor=new ReportProcessor(service);
|
2
|
Thread senderThread=new Thread(processor);
|
23.启动这3个线程。
1
|
System.out.printf("Main: Starting the Threads\n");
|
2
|
faceThread.start();
|
3
|
onlineThread.start();
|
4
|
senderThread.start();
|
24.等待ReportRequest线程的结束。
1
|
try {
|
2
|
System.out.printf("Main: Waiting for the report
|
3
|
generators.\n");
|
4
|
faceThread.join();
|
5
|
onlineThread.join();
|
6
|
} catch (InterruptedException e) {
|
7
|
e.printStackTrace();
|
8
|
}
|
25.使用shutdown()方法关闭执行者，使用awaitTermination()方法等待任务的结果。
1
|
System.out.printf("Main: Shutting down the executor.\n");
|
2
|
executor.shutdown();
|
3
|
try {
|
4
|
executor.awaitTermination(1, TimeUnit.DAYS);
|
5
|
} catch (InterruptedException e) {
|
6
|
e.printStackTrace();
|
7
|
}
|
26.设置ReportSender对象的end属性值为true，结束它的执行。
1
|
processor.setEnd(true);
|
2
|
System.out.println("Main: Ends");
|
这是如何工作的…
在示例的主类中，你使用Executors类的newCachedThreadPool()方法创建ThreadPoolExecutor。然后，使用这个对象初始化一个CompletionService对象，因为CompletionService需要使用一个执行者来执行任务。利用CompletionService执行一个任务，你需要使用submit()方法，如在ReportRequest类中。
当其中一个任务被执行，CompletionService完成这个任务的执行时，这个CompletionService在一个队列中存储Future对象来控制它的执行。poll()方法用来查看这个列队，如果有任何任务执行完成，那么返回列队的第一个元素，它是一个已完成任务的Future对象。当poll()方法返回一个Future对象时，它将这个Future对象从队列中删除。这种情况下，你可以传两个属性给那个方法，表明你想要等任务结果的时间，以防队列中的已完成任务的结果是空的。
一旦CompletionService对象被创建，你创建2个ReportRequest对象，用来执行3个ReportGenerator任务，每个都在CompletionService中，和一个ReportSender任务，它将会处理已提交给2个ReportRequest对象的任务所产生的结果。
不止这些…
CompletionService类可以执行Callable和Runnable任务。在这个示例中，你已经使用Callable，但你同样可以提交Runnable对象。由于Runnable对象不会产生结果，CompletionService类的理念不适用于这些情况。
这个类同样提供其他两个方法，用来获取已完成任务的Future对象。这两个方法如下：
- poll()：不带参数版本的poll()方法，检查是否有任何Future对象在队列中。如果列队是空的，它立即返回null。否则，它返回第一个元素，并从列队中删除它。
- take()：这个方法，不带参数。检查是否有任何Future对象在队列中。如果队列是空的，它阻塞线程直到队列有一个元素。当队列有元素，它返回第一元素，并从列队中删除它。
参见
- 在第4章，线程执行者中的执行者执行返回结果的任务指南
线程执行者（十一）执行者分离任务的启动和结果的处理相关推荐
- Elasticsearch源码分析—线程池(十一) ——就是从队列里处理请求
Elasticsearch源码分析-线程池(十一) 转自:https://www.felayman.com/articles/2017/11/10/1510291570687.html 线程池 每个节 ...
- java线程开启不了_Java中多线程启动，为什么调用的是start方法，而不是run方法？...
前言 大年初二,大家新年快乐,我又开始码字了.写这篇文章,源于在家和基友交流的时候,基友问到了,我猛然发现还真是这么回事,多线程启动调用的都是start,那么为什么没人掉用run呢?于是打开我的ide ...
- 线程的状态：分离（detached）和joinable(可结合的）
线程分离 在任意一个时间点上,线程是可结合(joinable)或者是可分离的(detached).一个可结合线程是可以被其他线程收回资源和杀关闭.在被回收之前,他的存储器资源(栈等)是不释放的.而对于 ...
- java线程不能重复_Java中多线程重复启动
标签: 在面试时候经常被问到多线程的相关问题: 今天在测试的时候发现下面的代码会抛出异常: java.lang.IllegalThreadStateException public static vo ...
- 【Docker】1、 前后端分离项目 下载启动运行
人人开源前后端分离项目下载与配置 文章目录 人人开源前后端分离项目下载与配置 前后端分离框架介绍 后端项目下载与配置 1.renren-fast后台项目介绍 2.开发环境搭建 3.下载后端renren ...
- java构造函数中启动线程_java-为什么不在构造函数中启动线程？ 如何终止？
java-为什么不在构造函数中启动线程? 如何终止? 我正在学习如何在Java中使用线程. 我写了一个实现Runnable的类,以同时运行到另一个线程. 主线程处理侦听串行端口,而第二个线程将处理向同 ...
- guns企业高级单体版（前后端不分离）运行启动
单体版分前后端分离与不分离,这里分享前后端不分离的搭建方法 访问guns官网https://www.stylefeng.cn,登录后可查看教程(账号密码见群公告) 官方教程不是最新的,有些地方写的不是 ...
- java怎么让main方法不退出_JAVA线程池原理源码解析—为什么启动一个线程池，提交一个任务后，Main方法不会退出？...
public static void main(String[] args) { ExecutorService service = Executors.newFixedThreadPool(10); ...
- C++版本OpenCv教程(十一)多通道分离与合并
在图像颜色模型中不同的分量存放在不同的通道中,如果我们只需要颜色模型的某一个分量,例如只需要处理RGB图像中的红色通道,可以将红色通道从三通道的数据中分离出来再进行处理,这种方式可以减少数据所占据的内 ...
最新文章
- python 开发版-MicroPython的开发板
- 微服务架构：动态配置中心搭建
- PostgreSQL 当有多个索引可选时，优化器如何选择
- 深度信念网络Deep Belief Networks资料汇总
- 如何在Django1.6结合Python3.4版本中使用MySql
- ORM中的Model与DDD中的DomainModel
- Spark入门(Python)
- python tableview 列宽_QTableView设置列宽
- php图片提交,php+js实现图片的上传、裁剪、预览、提交示例
- 使用Python+md5删除本地重复(同一张不重名)的照片
- Qt QTimer在线程的应用与思考
- RDA实现SQL CE与SQL Server间数据存取
- GWR4 软件输入数据制作
- windows server 2008 R2 SP1多国语言包官方下载
- 整车控制器（VCU，vehicle Controller Unit）
- MAC OS搭建pyhton+selenium+pycharm实现web自动化测试
- Java Web 开发后续(四)
- python镜像安装re模块_Python模块之re模块
- 怎样给计算机桌面设密码,怎样给电脑设置锁屏密码
- F - Color the ball