CompletionService的功能是以异步的方式一边生产新的任务，一边处理已完成任务的结果，这样可以将执行任务与处理任务分离开来进行处理。今天我们通过实例来学习一下CompletionService的用法。
CompletionService的简单使用
使用submit()方法执行任务，使用take取得已完成的任务，并按照完成这些任务的时间顺序处理它们的结果。
一、CompletionService的submit方法
public class CompletionServiceTest {
public static void main(String[] args) throws Exception {
ExecutorService service = Executors.newFixedThreadPool(5);
CompletionService<String> completionService = new ExecutorCompletionService<String>(service);
for (int i = 0; i < 5; i++) {
completionService.submit(new ReturnAfterSleepCallable(i));
}
System.out.println("after submit");
for (int i = 0; i < 5; i++) {
System.out.println("result: " + completionService.take().get()); // 这个方法是阻塞的
}
System.out.println("after get");
service.shutdown();
}
private static class ReturnAfterSleepCallable implements Callable<String> {
int sleep;
public ReturnAfterSleepCallable(int sleep) {
this.sleep = sleep;
}
@Override
public String call() throws Exception {
TimeUnit.SECONDS.sleep(sleep);
return System.currentTimeMillis() + ",sleep=" + String.valueOf(sleep);
}
}
}
运行的结果如下：
after submit
result: ,sleep=
result: ,sleep=
result: ,sleep=
result: ,sleep=
result: ,sleep=
after get
官方文档上的说明：
Submits a value-returning task for execution and returns a Future representing the pending results of the task. Upon completion, this task may be taken or polled.
二、使用CompletionService的take方法
take()方法取得最先完成任务的Future对象，谁执行时间最短谁最先返回。
package com.linux.thread; import java.util.Random;
import java.util.concurrent.*; public class RunMain1 {
public static void main(String[] args) {
try {
ExecutorService executorService = Executors.newCachedThreadPool();
ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executorService);
for (int i = 0; i < 10; i++) {
completionService.submit(new Callable<String>() {
@Override
public String call() throws Exception {
int sleepValue = new Random().nextInt(5);
System.out.println("sleep = " + sleepValue + ", name: " + Thread.currentThread().getName());
TimeUnit.SECONDS.sleep(sleepValue);
return "huhx: " + sleepValue + ", " + Thread.currentThread().getName();
}
});
}
for (int i = 0; i < 10; i++) {
System.out.println(completionService.take().get());
}
executorService.shutdown();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
一次的运行结果如下：
sleep = , name: pool--thread-
sleep = , name: pool--thread-
sleep = , name: pool--thread-
huhx: , pool--thread-
huhx: , pool--thread-
sleep = , name: pool--thread-
sleep = , name: pool--thread-
sleep = , name: pool--thread-
sleep = , name: pool--thread-
sleep = , name: pool--thread-
huhx: , pool--thread-
sleep = , name: pool--thread-
sleep = , name: pool--thread-
huhx: , pool--thread-
huhx: , pool--thread-
huhx: , pool--thread-
huhx: , pool--thread-
huhx: , pool--thread-
huhx: , pool--thread-
huhx: , pool--thread-
官方文档上的说明：
Retrieves and removes the Future representing the next completed task, waiting if none are yet present.
三、使用CompletionService的poll方法
方法poll的作用是获取并移除表示下一个已完成任务的Future，如果不存在这样的任务，则返回null，方法poll是无阻塞的。
package com.linux.thread;
import java.util.concurrent.*;
public class RunMain2 {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
CompletionService<String> service = new ExecutorCompletionService<String>(executorService);
service.submit(new Callable<String>() {
@Override
public String call() throws Exception {
TimeUnit.SECONDS.sleep(3);
System.out.println("3 seconds pass.");
return "3秒";
}
});
System.out.println(service.poll());
executorService.shutdown();
}
}
运行的结果如下：
null
seconds pass.
官方文档上的说明：
Retrieves and removes the Future representing the next completed task or null if none are present.