线程执行者(十一)执行者分离任务的启动和结果的处理

简介:
声明:本文是《 Java 7 Concurrency Cookbook 》的第四章,作者: Javier Fernández González     译者:许巧辉     校对:方腾飞,叶磊

执行者分离任务的启动和结果的处理

通常,当你使用执行者执行并发任务时,你将会提交 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章,线程执行者中的执行者执行返回结果的任务指南
文章转自  并发编程网-ifeve.com
目录
相关文章
|
1月前
|
缓存 负载均衡 Java
c++写高性能的任务流线程池(万字详解!)
本文介绍了一种高性能的任务流线程池设计,涵盖多种优化机制。首先介绍了Work Steal机制,通过任务偷窃提高资源利用率。接着讨论了优先级任务,使不同优先级的任务得到合理调度。然后提出了缓存机制,通过环形缓存队列提升程序负载能力。Local Thread机制则通过预先创建线程减少创建和销毁线程的开销。Lock Free机制进一步减少了锁的竞争。容量动态调整机制根据任务负载动态调整线程数量。批量处理机制提高了任务处理效率。此外,还介绍了负载均衡、避免等待、预测优化、减少复制等策略。最后,任务组的设计便于管理和复用多任务。整体设计旨在提升线程池的性能和稳定性。
72 5
|
3月前
|
前端开发 JavaScript 大数据
React与Web Workers:开启前端多线程时代的钥匙——深入探索计算密集型任务的优化策略与最佳实践
【8月更文挑战第31天】随着Web应用复杂性的提升,单线程JavaScript已难以胜任高计算量任务。Web Workers通过多线程编程解决了这一问题,使耗时任务独立运行而不阻塞主线程。结合React的组件化与虚拟DOM优势,可将大数据处理等任务交由Web Workers完成,确保UI流畅。最佳实践包括定义清晰接口、加强错误处理及合理评估任务特性。这一结合不仅提升了用户体验,更为前端开发带来多线程时代的全新可能。
67 1
|
3月前
|
存储 监控 Java
|
4月前
|
Java Linux
Java演进问题之1:1线程模型对于I/O密集型任务如何解决
Java演进问题之1:1线程模型对于I/O密集型任务如何解决
|
3月前
|
Cloud Native Java 调度
项目环境测试问题之线程同步器会造成执行完任务的worker等待的情况如何解决
项目环境测试问题之线程同步器会造成执行完任务的worker等待的情况如何解决
|
3月前
|
Java 测试技术 PHP
父子任务使用不当线程池死锁怎么解决?
在Java多线程编程中,线程池有助于提升性能与资源利用效率,但若父子任务共用同一池,则可能诱发死锁。本文通过一个具体案例剖析此问题:在一个固定大小为2的线程池中,父任务直接调用`outerTask`,而`outerTask`再次使用同一线程池异步调用`innerTask`。理论上,任务应迅速完成,但实际上却超时未完成。经由`jstack`输出的线程调用栈分析发现,线程陷入等待状态,形成“死锁”。原因是子任务需待父任务完成,而父任务则需等待子任务执行完毕以释放线程,从而相互阻塞。此问题在测试环境中不易显现,常在生产环境下高并发时爆发,重启或扩容仅能暂时缓解。
|
4月前
|
安全
线程操纵术并行策略问题之ForkJoinTask提交任务的问题如何解决
线程操纵术并行策略问题之ForkJoinTask提交任务的问题如何解决
|
4月前
线程操纵术并行策略问题之ForkJoinTask提交任务的问题如何解决
线程操纵术并行策略问题之ForkJoinTask提交任务的问题如何解决
|
4月前
|
并行计算 安全 Java
线程操纵术并行策略问题之任务执行器(Executor)问题如何解决
线程操纵术并行策略问题之任务执行器(Executor)问题如何解决
|
4月前
|
设计模式 安全 Java
Java面试题:如何实现一个线程安全的单例模式,并确保其在高并发环境下的内存管理效率?如何使用CyclicBarrier来实现一个多阶段的数据处理任务,确保所有阶段的数据一致性?
Java面试题:如何实现一个线程安全的单例模式,并确保其在高并发环境下的内存管理效率?如何使用CyclicBarrier来实现一个多阶段的数据处理任务,确保所有阶段的数据一致性?
57 0