CompletionService小记

CompletionService小记

在使用ExecutorService时，经过sumit执行一个Callable的时候，会当即返回一个异步任务结果，而后经过get获取异步任务结果的时候会阻塞，以下面这种状况，代码以下，

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * Created by xinxingegeya on 16/3/22.
 */
public class ExecutorServiceTest {

    public static void main(String args[])
            throws ExecutionException, InterruptedException {

        ExecutorService executor = Executors.newCachedThreadPool();

        List<Future<Integer>> futureList = new ArrayList<>();
        for (int i = 5; i > 0; i--) {
            final int taskID = i;
            //CompletionService.submit()
            Future<Integer> res = executor.submit(new Callable<Integer>() {
                public Integer call() throws Exception {
                    Thread.sleep(taskID * 1000);
                    return taskID;
                }
            });
            futureList.add(res);
        }

        for (Future<Integer> future : futureList) {
            int result = future.get();
            System.out.println(result);
        }

        executor.shutdown();

        while (executor.isTerminated()) {
            executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
        }
    }
}

经过代码能够看到，在遍历任务结果集的时候，获取第一个任务结果，而后get会阻塞，会阻塞大概五秒的时间，而对于后面的任务，可能已经执行完成了，但就是阻塞在了第一Future上，这是不合理的，那么如何改进，在这里应该要使用CompletionService，以下代码，

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.TimeUnit;

/**
 * Created by xinxingegeya on 16/3/22.
 */
public class ExecutorCompletionServiceTest {

    public static void main(String args[]) throws InterruptedException {

        ExecutorService executor = Executors.newCachedThreadPool();

        CompletionService<Integer> cs = new ExecutorCompletionService<>(
            executor);
        for (int i = 5; i > 0; i--) {
            final int taskID = i;
            //CompletionService.submit()
            cs.submit(new Callable<Integer>() {
                public Integer call() throws Exception {
                    Thread.sleep(taskID * 1000);
                    return taskID;
                }
            });
        }
        // 可能作一些事情
        for (int i = 1; i < 5; i++) {
            try {
                int result = cs.take().get();//CompletionService.take()返回Future
                System.out.println(result);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

        executor.shutdown();

        while (executor.isTerminated()) {
            executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
        }

        System.out.println("done");
    }
}

使用CompletionService的好处就是优先获取已经执行完成的任务结果，而不会一味的等待尚未执行完成的任务。

CompletionService的实现机制就是利用BlockingQueue，若是任务已经执行完成，会把Future放进阻塞队列里面，若是阻塞队列中有已经完成的任务，那么就取出来（take）获取异步任务的结果。

==============END==============