Java中的CompletionService批量異步執(zhí)行詳解

更新時間：2023年12月22日 09:05:54 作者：Java面試365

這篇文章主要介紹了Java中的CompletionService批量異步執(zhí)行詳解,我們知道線程池可以執(zhí)行異步任務,同時可以通過返回值Future獲取返回值,所以異步任務大多數采用ThreadPoolExecutor+Future,需要的朋友可以參考下

前景引入

我們知道線程池可以執(zhí)行異步任務，同時可以通過返回值Future獲取返回值，所以異步任務大多數采用ThreadPoolExecutor+Future，如果存在如下情況，需要從任務一二三中獲取返回值后，保存到數據庫中，用異步邏輯實現(xiàn)代碼應該如下所示。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    Future<Integer> f1 = executorService.submit(() -> {
        System.out.println("執(zhí)行任務一");
        return 1;
    });
    Future<Integer> f2 = executorService.submit(() -> {
        System.out.println("執(zhí)行任務二");
        return 2;
    });
    Future<Integer> f3 = executorService.submit(() -> {
        System.out.println("執(zhí)行任務三");
        return 3;
    });
    Integer r1 = f1.get();
    executorService.execute(()->{
        // 省略保存r1操作
        System.out.println(r1);
    });
    Integer r2 = f2.get();
    executorService.execute(()->{
        // 省略保存r2操作
        System.out.println(r2);
    });
    Integer r3 = f3.get();
    executorService.execute(()->{
        // 省略保存r3操作
        System.out.println(r3);
    });
    executorService.shutdown();
}

這樣寫的代碼一點毛病沒有，邏輯都是正常的，但如果存在任務一查詢了比較耗時的操作，由于f1.get是阻塞執(zhí)行，那么就算任務二和任務三已經返回結果，任務二的返回值和任務三的返回值都是不能保存到數據庫的，因為f1.get將主線程阻塞了。

批量異步實現(xiàn)

那可以如何處理呢？可以采用萬能的阻塞隊列，任務先執(zhí)行完畢的先入隊，這樣可以保證其它線程入庫的速度不受影響，提高效率。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue(3);
    Future<Integer> f1 = executorService.submit(() -> {
        System.out.println("執(zhí)行任務一");
        Thread.sleep(5000);
        return 1;
    });
    Future<Integer> f2 = executorService.submit(() -> {
        System.out.println("執(zhí)行任務二");
        return 2;
    });
    Future<Integer> f3 = executorService.submit(() -> {
        System.out.println("執(zhí)行任務三");
        Thread.sleep(3000);
        return 3;
    });
    executorService.execute(()->{
        try {
            Integer r1 = f1.get();
            // 阻塞隊列入隊操作
            queue.put(r1);
            System.out.println(r1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
    executorService.execute(()->{
        try {
            Integer r2 = f2.get();
            queue.put(r2);
            System.out.println(r2);
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
    executorService.execute(()->{
        try {
            Integer r3 = f3.get();
            queue.put(r3);
            System.out.println(r3);
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
    // 循環(huán)次數不要使用queue.size限制，因為不同時刻queue.size值是有可能不同的
    for (int i = 0; i <3; i++) {
        Integer integer = queue.take();
        // 省略保存integer操作
        executorService.execute(()->{
            System.out.println("保存入庫=="+integer);
        });
    }
    executorService.shutdown();
}

產生結果如下

同樣的在生產中不建議使用，因為SDK為我們提供了工具類CompletionService，CompletionService內部就維護了一個阻塞隊列，唯一與上述代碼實現(xiàn)有所區(qū)別的是，阻塞隊列入庫的是Future對象，其余原理類似。

CompletionService

如何創(chuàng)建CompletionService

CompletionService同樣是一個接口，其具體實現(xiàn)為ExecutorCompletionService，創(chuàng)建CompletionService對象有兩種方式

public ExecutorCompletionService(Executor executor);
public ExecutorCompletionService(Executor executor,BlockingQueue<Future<V>> completionQueue)

CompletionService對象的創(chuàng)建都是需要指定線程池，如果在創(chuàng)建時沒有傳入阻塞對象，那么會采用默認的LinkedBlockingQueue無界阻塞隊列，如果應用到生產可能會產生OOM的情況，這是需要注意的。

CompletionService初體驗

CompletionService如何做到批量執(zhí)行異步任務呢，將上述場景采用CompletionService實現(xiàn)下

public static void main(String[] args) throws InterruptedException, ExecutionException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    CompletionService completionService = new ExecutorCompletionService(executorService);
    Future<Integer> f1 = completionService.submit(() -> {
        System.out.println("執(zhí)行任務一");
        Thread.sleep(5000);
        return 1;
    });
    Future<Integer> f2 = completionService.submit(() -> {
        System.out.println("執(zhí)行任務二");
        return 2;
    });
    Future<Integer> f3 = completionService.submit(() -> {
        System.out.println("執(zhí)行任務三");
        Thread.sleep(3000);
        return 3;
    });
    for (int i = 0; i <3 ; i++) {
        Future take = completionService.take();
        Integer integer = (Integer) take.get();
        executorService.execute(()->{
            System.out.println("執(zhí)行入庫=="+integer);
        });
    }
    executorService.shutdown();
}

CompletionService接口說明

CompletionService的方法不多，使用起來比較簡單，方法簽名如下

public static void main(String[] args) throws InterruptedException, ExecutionException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    CompletionService completionService = new ExecutorCompletionService(executorService);
    Future<Integer> f1 = completionService.submit(() -> {
        System.out.println("執(zhí)行任務一");
        Thread.sleep(5000);
        return 1;
    });
    Future<Integer> f2 = completionService.submit(() -> {
        System.out.println("執(zhí)行任務二");
        return 2;
    });
    Future<Integer> f3 = completionService.submit(() -> {
        System.out.println("執(zhí)行任務三");
        Thread.sleep(3000);
        return 3;
    });
    for (int i = 0; i <3 ; i++) {
        Future take = completionService.take();
        Integer integer = (Integer) take.get();
        executorService.execute(()->{
            System.out.println("執(zhí)行入庫=="+integer);
        });
    }
    executorService.shutdown();
}