Java批处理ExecutorService/CompletionService

服务端接收一个请求，常常需要同时进行几个计算或者向其他服务发送请求，最后拼装结果返回上游。本文就来看下JDK提供几个并行处理方案，牵涉到ExcecutorService/CompletionService。要实现的场景是请求有超时限制，如果所有操作都计算完成，则全部拼装返回；否则只拼装部分完成的结果。

1.前提

//任务类，sleep一个时间代表这个计算需要的耗时，返回一个计算结果。

public class MyTask implements Callable<Integer> {

    private int id;

    private int time;

    public MyTask(int i, int time) {

        this.id = i;

        this.time = time;

    }

    @Override

    public Integer call() throws Exception {

        Thread.sleep(time);

        return id;

    }

}

//线程池

ExecutorService threadPool = new ThreadPoolExecutor(10, 20,

60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100));

2.任务逐个等待

提交任务后，逐个等待结果返回。

//总共150ms超时

List<Future<Integer>> futures = new ArrayList<Future<Integer>>();

futures.add(threadPool.submit(new MyTask(1, 60)));

futures.add(threadPool.submit(new MyTask(2, 150)));

Integer res = 0;

for (int i = 0; i < futures.size(); ++i) {

try {

Integer tmp = futures.get(i).get(50 + i * 50, TimeUnit.MILLESECONDS);

res += tmp;

} catch (Exception e) {

//nothing

}

}

System.out.println(res);

打印结果为0，实际上第60ms时第一个计算已经完成，本可以返回第一个结果的。这是因为生产者和消费者一一对应，调度不当。

3.等待解耦

用一个队列来实现生产者和消费者解耦，生产者把结果放到一个队列中，消费者从队列中取结果，不按照任务提交顺序等待，只要有结果就会消耗。CompletionService就是对Executor和异步队列的封装。

CompletionService<Integer> service = new ExecutorCompletionService<Integer>(threadPool);

List<Future<Integer>> futures = new ArrayList<Future<Integer>>();

futures.add(service.submit(new MyTask(1, 60)));

futures.add(service.submit(new MyTask(2, 150)));

List<Integer> result = new ArrayList<Integer>(futures.size());

for (int i = 0; i < futures.size(); ++i) {

    Future<Integer> future = service.poll(50 + i * 50, TimeUnit.MILLISECONDS);

    if (null != future){

        result.add(future.get());

    }

}

int res = 0;

for (Integer i : result) {

    res += null == i ? 0 : i;

}

System.out.println(res);

打印结果为1，即第一个计算的结果。谁先完成就取谁。

4.充分利用时间

前面两个方案为每个任务都设置了一个固定的等待时间，两者之和不超过超时限制。这没有充分利用时间，两个任务是并发互不影响的，其各自可利用的时间应该是超时时间，而不应该是两者之和为超时时间。如果第一个任务耗时140，第二个耗时60，前两个任务就只能获得第二个任务的结果。如果两者都设置超时为150，就能获得2个结果。

List<MyTask> tasks = new ArrayList<MyTask>();

tasks.add(new MyTask(1, 140));

tasks.add(new MyTask(2, 60));

List<Future<Integer>> futures = threadPool.invokeAll(tasks, 150, TimeUnit.MILLISECONDS);

int res = 0;

for (Future<Integer> future : futures) {

    System.out.println("isDone:" + future.isDone());

    System.out.println("isCancel:" + future.isCancelled());

    if (future.isCancelled()) {

        continue;

    }

    res += future.get();

}

System.out.println(res);

打印结果是3。上述方法利用ExecutorService的invokeAll方法，该方法在所有任务都完成或等待超时时返回，超时的时候会取消还没有完成的任务，通过判断任务是否被取消来判断任务是否计算完成。

该文章说的是结果互不影响、有结果就能返回的应用场景。如果结果需要按照先后顺序进行合并，或者只需要等待一个计算完成，就不一定需要方案3了。

巴特西

Java批处理ExecutorService/CompletionService

1.前提

2.任务逐个等待

3.等待解耦

4.充分利用时间

最新文章

热门文章