Java线程之CompletionService批处理任务
2016-05-05 20:36
465 查看
如果你向Executor提交了一个批处理任务,并且希望在它们完成后获得结果,怎么办呢?
为此你可以保存与每个任务相关联的Future,然后不断地调用 timeout为零的get,来检验Future是否完成。这样做固然可以,但却相当乏味。幸运的是,还有一个更好的方法:完成服务 (Completion service)。
CompletionService整合了Executor和BlockingQueue的功能。
你可以将Callable任务提交给它去执行,然 后使用类似于队列中的take和poll方法,在结果完整可用时获得这个结果,像一个打包的Future。 ExecutorCompletionService是实现CompletionService接口的一个类,并将计算任务委托给一个Executor。
ExecutorCompletionService的实现相当直观。它在构造函数中创建一个BlockingQueue,用它去保持完成的结果。 计算完成时会调用FutureTask中的done方法。
当提交一个任务后,首先把这个任务包装为一个QueueingFuture,它是 FutureTask的一个子类,然后覆写done方法,将结果置入BlockingQueue中,take和poll方法委托给了 BlockingQueue,它会在结果不可用时阻塞。
直接看demo:
package javademo;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
/***
* 两钟方式出来线程运行结果
* @author think
*
*/
public class CompletionServiceTest {
public static void main(String[] args) throws Exception {
CompletionServiceTest cst = new CompletionServiceTest();
cst.count1();
cst.count2();
}
/***
* 使用阻塞容器保存每次Executor处理的结果,在后面进行统一处理
* @throws Exception
*/
public void count1() throws Exception{
ExecutorService exec = Executors.newCachedThreadPool();
BlockingQueue<Future<Integer>> queue = new LinkedBlockingQueue<Future<Integer>>();
for(int i=0; i<10; i++){
Future<Integer> future =exec.submit(getTask());
queue.add(future);
}
int sum = 0;
int queueSize = queue.size();
for(int i=0; i<queueSize; i++){
sum += queue.take().get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
/***
* 使用CompletionService(完成服务)保持Executor处理的结果
* @throws InterruptedException
* @throws ExecutionException
*/
public void count2() throws InterruptedException, ExecutionException{
ExecutorService exec = Executors.newCachedThreadPool();
CompletionService<Integer> execcomp = new ExecutorCompletionService<Integer>(exec);
for(int i=0; i<10; i++){
execcomp.submit(getTask());
}
int sum = 0;
for(int i=0; i<10; i++){
//检索并移除表示下一个已完成任务的 Future,如果目前不存在这样的任务,则等待。
Future<Integer> future = execcomp.take();
sum += future.get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
/**
* 得到一个任务
* @return Callable
*/
public Callable<Integer> getTask(){
final Random rand = new Random();
Callable<Integer> task = new Callable<Integer>(){
@Override
public Integer call() throws Exception {
int i = rand.nextInt(10);
int j = rand.nextInt(10);
int sum = i*j;
System.out.print(sum+"\t");
return sum;
}
};
return task;
}
}
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
public class Test17 {
public static void main(String[] args) throws Exception {
Test17 t = new Test17();
t.count1();
t.count2();
}
//使用阻塞容器保存每次Executor处理的结果,在后面进行统一处理
public void count1() throws Exception{
ExecutorService exec = Executors.newCachedThreadPool();
BlockingQueue<Future<Integer>> queue = new LinkedBlockingQueue<Future<Integer>>();
for(int i=0; i<10; i++){
Future<Integer> future =exec.submit(getTask());
queue.add(future);
}
int sum = 0;
int queueSize = queue.size();
for(int i=0; i<queueSize; i++){
sum += queue.take().get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
//使用CompletionService(完成服务)保持Executor处理的结果
public void count2() throws InterruptedException, ExecutionException{
ExecutorService exec = Executors.newCachedThreadPool();
CompletionService<Integer> execcomp = new ExecutorCompletionService<Integer>(exec);
for(int i=0; i<10; i++){
execcomp.submit(getTask());
}
int sum = 0;
for(int i=0; i<10; i++){
//检索并移除表示下一个已完成任务的 Future,如果目前不存在这样的任务,则等待。
Future<Integer> future = execcomp.take();
sum += future.get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
//得到一个任务
public Callable<Integer> getTask(){
final Random rand = new Random();
Callable<Integer> task = new Callable<Integer>(){
@Override
public Integer call() throws Exception {
int i = rand.nextInt(10);
int j = rand.nextInt(10);
int sum = i*j;
System.out.print(sum+"\t");
return sum;
}
};
return task;
}
/**
* 执行结果:
6 6 14 40 40 0 4 7 0 0 总数为:106
12 6 12 54 81 18 14 35 45 35 总数为:312
*/
}
为此你可以保存与每个任务相关联的Future,然后不断地调用 timeout为零的get,来检验Future是否完成。这样做固然可以,但却相当乏味。幸运的是,还有一个更好的方法:完成服务 (Completion service)。
CompletionService整合了Executor和BlockingQueue的功能。
你可以将Callable任务提交给它去执行,然 后使用类似于队列中的take和poll方法,在结果完整可用时获得这个结果,像一个打包的Future。 ExecutorCompletionService是实现CompletionService接口的一个类,并将计算任务委托给一个Executor。
ExecutorCompletionService的实现相当直观。它在构造函数中创建一个BlockingQueue,用它去保持完成的结果。 计算完成时会调用FutureTask中的done方法。
当提交一个任务后,首先把这个任务包装为一个QueueingFuture,它是 FutureTask的一个子类,然后覆写done方法,将结果置入BlockingQueue中,take和poll方法委托给了 BlockingQueue,它会在结果不可用时阻塞。
直接看demo:
package javademo;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
/***
* 两钟方式出来线程运行结果
* @author think
*
*/
public class CompletionServiceTest {
public static void main(String[] args) throws Exception {
CompletionServiceTest cst = new CompletionServiceTest();
cst.count1();
cst.count2();
}
/***
* 使用阻塞容器保存每次Executor处理的结果,在后面进行统一处理
* @throws Exception
*/
public void count1() throws Exception{
ExecutorService exec = Executors.newCachedThreadPool();
BlockingQueue<Future<Integer>> queue = new LinkedBlockingQueue<Future<Integer>>();
for(int i=0; i<10; i++){
Future<Integer> future =exec.submit(getTask());
queue.add(future);
}
int sum = 0;
int queueSize = queue.size();
for(int i=0; i<queueSize; i++){
sum += queue.take().get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
/***
* 使用CompletionService(完成服务)保持Executor处理的结果
* @throws InterruptedException
* @throws ExecutionException
*/
public void count2() throws InterruptedException, ExecutionException{
ExecutorService exec = Executors.newCachedThreadPool();
CompletionService<Integer> execcomp = new ExecutorCompletionService<Integer>(exec);
for(int i=0; i<10; i++){
execcomp.submit(getTask());
}
int sum = 0;
for(int i=0; i<10; i++){
//检索并移除表示下一个已完成任务的 Future,如果目前不存在这样的任务,则等待。
Future<Integer> future = execcomp.take();
sum += future.get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
/**
* 得到一个任务
* @return Callable
*/
public Callable<Integer> getTask(){
final Random rand = new Random();
Callable<Integer> task = new Callable<Integer>(){
@Override
public Integer call() throws Exception {
int i = rand.nextInt(10);
int j = rand.nextInt(10);
int sum = i*j;
System.out.print(sum+"\t");
return sum;
}
};
return task;
}
}
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
public class Test17 {
public static void main(String[] args) throws Exception {
Test17 t = new Test17();
t.count1();
t.count2();
}
//使用阻塞容器保存每次Executor处理的结果,在后面进行统一处理
public void count1() throws Exception{
ExecutorService exec = Executors.newCachedThreadPool();
BlockingQueue<Future<Integer>> queue = new LinkedBlockingQueue<Future<Integer>>();
for(int i=0; i<10; i++){
Future<Integer> future =exec.submit(getTask());
queue.add(future);
}
int sum = 0;
int queueSize = queue.size();
for(int i=0; i<queueSize; i++){
sum += queue.take().get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
//使用CompletionService(完成服务)保持Executor处理的结果
public void count2() throws InterruptedException, ExecutionException{
ExecutorService exec = Executors.newCachedThreadPool();
CompletionService<Integer> execcomp = new ExecutorCompletionService<Integer>(exec);
for(int i=0; i<10; i++){
execcomp.submit(getTask());
}
int sum = 0;
for(int i=0; i<10; i++){
//检索并移除表示下一个已完成任务的 Future,如果目前不存在这样的任务,则等待。
Future<Integer> future = execcomp.take();
sum += future.get();
}
System.out.println("总数为:"+sum);
exec.shutdown();
}
//得到一个任务
public Callable<Integer> getTask(){
final Random rand = new Random();
Callable<Integer> task = new Callable<Integer>(){
@Override
public Integer call() throws Exception {
int i = rand.nextInt(10);
int j = rand.nextInt(10);
int sum = i*j;
System.out.print(sum+"\t");
return sum;
}
};
return task;
}
/**
* 执行结果:
6 6 14 40 40 0 4 7 0 0 总数为:106
12 6 12 54 81 18 14 35 45 35 总数为:312
*/
}
相关文章推荐
- Myeclipse的使用
- struts.properties配置详解
- 使用纯java基本语言实现泰勒展开的sin和cos计算
- 排序算法java 一 --快速排序、直接插入、希尔排序
- Eclipse常用快捷键
- JAVA 内存泄露详解(原因、例子及解决)
- Java并发编程的艺术(上)
- JAVA修饰符
- Java垃圾回收
- 非spring组件servlet、filter、interceptor中注入spring bean
- java软件设计的三层结构
- java序列化ClassNotFoundException
- java文件输入输出流
- Java实现十进制数转化为二进制数的算法
- Java设计模式-适配器模式Adapter
- 【java】java.util.Stack类
- java的集合类API
- Java:单例模式的七种写法[转]
- Java强引用、软引用、弱引用和虚引用
- 一个展现java多线程原理的例子