Java线程之CompletionService

Posted 2020-06-14

CompletionService的使用

　　当我们使用ExecutorService创建一个线程池时, 如果执行了多个Callable任务后, 每个Callable任务都会产生一个Future, 如果我们需要处理这些任务产生的结果, 那么就需要将这些Future放入一个线性表中, 用于之后的数据处理.

　　有了CompletionService, 我们就不用人为地去创建线性表来存这些Future了, CompleService是一个更高级的EexcutorService, 它自身自带一个线程安全的线性表, 无需用户额外创建. 它提供了两个方法从线性表中取出结果:  poll()是非阻塞的, 若目前无结果, 则返回null, 当前线程继续运行不阻塞; take()是阻塞的, 若当前无结果, 则当前线程阻塞, 直到产生一个结果, 被取出返回, 当前线程继续运行.

代码如下:

import java.util.concurrent.*;

public class CompletionServiceTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(executorService);

        for (int i=0; i<5; i++) {
            completionService.submit(new Task());
        }
        executorService.shutdown();
        int count=0, index=1;
        while (count < 5) {
            Future<Integer> result = completionService.poll();
            if (result ==  null) {
                System.out.println(index + " 没有发现有完成的任务");
            }else {
                System.out.println(index + " 产生了一个随机数 " + result.get() );
                count++;
            }
            index++;
            TimeUnit.SECONDS.sleep(2);
        }
    }
}

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

public class Task implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        Random random = new Random();
        TimeUnit.SECONDS.sleep(1);
        return random.nextInt();
    }
}

 运行结果如下:

1 没有发现有完成的任务
2 产生了一个随机数 1068729946
3 产生了一个随机数 -754072428
4 产生了一个随机数 -711120809
5 产生了一个随机数 -1696523068
6 产生了一个随机数 1409817456