官方文档：https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html

1 示例

Demo1:验证CompletableFuture的阻塞和异步

  1. public class CompletableFutureTest {
  2.     public static void main(String[] args) {
  3.         Long start = System.currentTimeMillis();
  4.      //   new CompletableFutureTest().futureBlock();
  5.         Long end = System.currentTimeMillis();
  6.      //   System.out.println("futhre阻塞耗时：" + (end - start));
  7.         Long start2 = System.currentTimeMillis();
  8.         new CompletableFutureTest().futhreSync();
  9.         Long end2 = System.currentTimeMillis();
  10.         System.out.println("futhre异步耗时：" + (end2 - start2));
  11.     }
  13.     public void futureBlock() {
  14.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10,
  15.                 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10000),
  16.                 new ThreadFactoryBuilder().setNameFormat("checkExcelImportProductInfoList-pool-%d").build(),
  17.                 new ThreadPoolExecutor.CallerRunsPolicy());
  18.         List<CompletableFuture> list = new ArrayList();
  19.         for (int i = 0; i < 10; i++) {
  20.             int finalI = i;
  21.             CompletableFuture future = CompletableFuture.supplyAsync(() -> this.toDo(finalI), threadPoolExecutor).exceptionally(e -> {
  22.                 e.printStackTrace();
  23.                 return null;
  24.             });
  25.             list.add(future);
  26.         }
  27.         list.stream().forEach(m -> {
  28.             try {
  29.                 Map map = (Map) m.get();//阻塞
  30.                 next(map);
  31.             } catch (InterruptedException e) {
  32.                 e.printStackTrace();
  33.             } catch (ExecutionException e) {
  34.                 e.printStackTrace();
  35.             }
  36.         });
  37.         threadPoolExecutor.shutdown();
  38.     }
  40.     public void futhreSync() {
  41.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10,
  42.                 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10000),
  43.                 new ThreadFactoryBuilder().setNameFormat("checkExcelImportProductInfoList-pool-%d").build(),
  44.                 new ThreadPoolExecutor.CallerRunsPolicy());
  45.         for (int i = 0; i < 10; i++) {
  46.             int finalI = i;
  47.             CompletableFuture future = CompletableFuture.supplyAsync(() -> this.toDo(finalI), threadPoolExecutor).thenApply(map -> {
  48.                 next(map);
  49.                 return null;
  50.             }).exceptionally(e -> {
  51.                 e.printStackTrace();
  52.                 return null;
  53.             });
  54.         }
  55.         threadPoolExecutor.shutdown();
  56.     }
  58.     public Map toDo(int i) {
  59.         if (i == 3 ) {
  60.             try {
  61.                 Thread.sleep(10000);
  62.             } catch (InterruptedException e) {
  63.                 e.printStackTrace();
  64.             }
  65.         }
  66.         if (i == 6 ) {
  67.             try {
  68.                 Thread.sleep(5000);
  69.             } catch (InterruptedException e) {
  70.                 e.printStackTrace();
  71.             }
  72.         }
  73.         Map map = new HashMap();
  74.         map.put(i, Thread.currentThread().getName());
  75.         return map;
  76.     }
  78.     public void next(Map map) {
  79.         Iterator i = map.keySet().iterator();
  80.         while (i.hasNext()) {
  81.             Integer integer = (Integer) i.next();
  82.             System.out.println("key:" + integer + " ;value:" + map.get(integer));
  83.         }
  84.     }
  86. }

运行阻塞方法futureBlock时，输出：

  1. key:0 ;value:checkExcelImportProductInfoList-pool-0
  2. key:1 ;value:checkExcelImportProductInfoList-pool-1
  3. key:2 ;value:checkExcelImportProductInfoList-pool-2
  4. key:3 ;value:checkExcelImportProductInfoList-pool-3
  5. key:4 ;value:checkExcelImportProductInfoList-pool-4
  6. key:5 ;value:checkExcelImportProductInfoList-pool-0
  7. key:6 ;value:checkExcelImportProductInfoList-pool-0
  8. key:7 ;value:checkExcelImportProductInfoList-pool-1
  9. key:8 ;value:checkExcelImportProductInfoList-pool-1
  10. key:9 ;value:checkExcelImportProductInfoList-pool-1
  11. futhre阻塞耗时：10082

在i=3时阻塞10s，导致其他线程无法进行下一步
image.png

运行异步futhreSync方法时，输出：

  1. key:0 ;value:checkExcelImportProductInfoList-pool-0
  2. key:1 ;value:checkExcelImportProductInfoList-pool-1
  3. futhre异步耗时：80
  4. key:5 ;value:checkExcelImportProductInfoList-pool-0
  5. key:7 ;value:checkExcelImportProductInfoList-pool-0
  6. key:8 ;value:checkExcelImportProductInfoList-pool-0
  7. key:9 ;value:checkExcelImportProductInfoList-pool-0
  8. key:2 ;value:checkExcelImportProductInfoList-pool-2
  9. key:4 ;value:checkExcelImportProductInfoList-pool-4
  10. key:6 ;value:checkExcelImportProductInfoList-pool-1
  11. key:3 ;value:checkExcelImportProductInfoList-pool-3

只有在i=3时阻塞10s，i=6时阻塞5s；
image.png

Demo2:验证CompletableFuture异步执行，并统计所有线程执行的结果

  1. package com.tools.threadDemo.CompletableFutureTest;
  3. import com.google.common.util.concurrent.ThreadFactoryBuilder;
  5. import java.util.*;
  6. import java.util.concurrent.*;
  8. public class CompletableFutureTest2 {
  9.     private ExecutorService threadPoolExecutor;
  11.     public static void main(String[] args) {
  12.         List<CompletableFuture> list = new ArrayList<>();
  13.         List<Map> resultList = new ArrayList<>();//模拟统计所有线程的结果
  14.         for (int i = 0; i < 10; i++) {
  15.             list.add(new CompletableFutureTest2().futhreSync(i, resultList));
  16.         }
  17.         CompletableFuture.allOf(list.toArray(new CompletableFuture[list.size()])).join();//待所有线程执行完成后再执行
  18.         System.out.println("result>>>>>>>>>>>>" + resultList);
  19.     }
  21.     private synchronized void initClientInfoExecutor(int corePoolSize, int maximumPoolSize) {
  22.         threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,
  23.                 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10000),
  24.                 new ThreadFactoryBuilder().setNameFormat("initClientInfoExecutor-pool-%d").build(),
  25.                 new ThreadPoolExecutor.CallerRunsPolicy());
  26.     }
  28.     public CompletableFuture futhreSync(int i, List<Map> mapList) {
  29.         initClientInfoExecutor(8, 10);
  30.         List<CompletableFuture> futureList = new ArrayList<>();
  32.         int finalI = i;
  33.         CompletableFuture future = CompletableFuture.supplyAsync(() -> this.toDo(finalI, mapList), threadPoolExecutor).thenApply(map -> {
  34.             next(map);
  35.             return map;
  36.         }).exceptionally(e -> {
  37.             e.printStackTrace();
  38.             return null;
  39.         });
  40.         threadPoolExecutor.shutdown();
  41.         return future;
  42.     }
  44.     /**
  45.      * 多线程执行的任务
  46.      */
  47.     public Map toDo(int i, List<Map> mapList) {
  48.         if (i == 3) {
  49.             try {
  50.                 Thread.sleep(10000);
  51.             } catch (InterruptedException e) {
  52.                 e.printStackTrace();
  53.             }
  54.         }
  55.         if (i == 6) {
  56.             try {
  57.                 Thread.sleep(5000);
  58.             } catch (InterruptedException e) {
  59.                 e.printStackTrace();
  60.             }
  61.         }
  63.         Map map = new HashMap();
  64.         map.put(i, Thread.currentThread().getName());
  65.         if (i % 2 == 0) {
  66.             Map map1 = new HashMap();
  67.             map1.put(i, Thread.currentThread().getName());
  68.             mapList.add(map1);
  70.         }
  71.         return map;
  72.     }
  74.     /**
  75.      * 必须依赖上一个线程执行完成后再执行
  76.      */
  77.     public void next(Map map) {
  78.         Iterator i = map.keySet().iterator();
  79.         while (i.hasNext()) {
  80.             Integer integer = (Integer) i.next();
  81.             System.out.println("key:" + integer + " ;value:" + map.get(integer));
  82.         }
  83.     }
  84. }

输出：

  1. key:0 ;value:initClientInfoExecutor-pool-0
  2. key:1 ;value:initClientInfoExecutor-pool-0
  3. key:2 ;value:initClientInfoExecutor-pool-0
  4. key:4 ;value:initClientInfoExecutor-pool-0
  5. key:5 ;value:initClientInfoExecutor-pool-0
  6. key:7 ;value:initClientInfoExecutor-pool-0
  7. key:8 ;value:initClientInfoExecutor-pool-0
  8. key:9 ;value:initClientInfoExecutor-pool-0
  9. key:6 ;value:initClientInfoExecutor-pool-0
  10. key:3 ;value:initClientInfoExecutor-pool-0
  11. result>>>>>>>>>>>>[{0=initClientInfoExecutor-pool-0}, {2=initClientInfoExecutor-pool-0}, {4=initClientInfoExecutor-pool-0}, {8=initClientInfoExecutor-pool-0}, {6=initClientInfoExecutor-pool-0}]

2 源码分析

2.1 andTree

andTree是allOf方法的底层实现，addTree的多个任务是各自独立并行执行，addTree将多个任务通过递归的方式两两组队，任一一个任务执行完成都会判断两两组队的两个任务是否都执行完了，如果是则触发上层的任务判断逻辑，直到最终所有任务都执行完了。

  1. public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
  2.         return andTree(cfs, 0, cfs.length - 1);
  3.     }
  1. //lo是起始数组索引，hi是终止数组索引
  2. static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs,
  3.                                            int lo, int hi) {
  4.         CompletableFuture<Void> d = new CompletableFuture<Void>();
  5.         if (lo > hi) // 空
  6.             d.result = NIL;
  7.         else {
  8.             CompletableFuture<?> a, b;
  9.             int mid = (lo + hi) >>> 1;
  10.              //通过递归，将多个任务转换成两两成对的，最底层的执行完了会触发上层
  11.             //这里的触发只是任务是否执行完成的判断逻辑，各任务是各自独立并行执行
  12.             if ((a = (lo == mid ? cfs[lo] :
  13.                       andTree(cfs, lo, mid))) == null ||
  14.                 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
  15.                       andTree(cfs, mid+1, hi)))  == null)
  16.                 throw new NullPointerException();
  17.             //如果a,b有一个未执行完，则返回false
  18.             if (!d.biRelay(a, b)) {
  19.         //不满足完成条件，生成一个中继并压栈，再次尝试同步完成。若不满足条件，ab任何一个完成后都会再间接调用它的tryFire。
  20.                 BiRelay<?,?> c = new BiRelay<>(d, a, b);
  21.                 //除非ab均完成，否则bipush要进ab两者的栈。
  22.                 a.bipush(b, c);
  23.                 c.tryFire(SYNC);
  24.             }
  25.         }
  26.         return d;
  27.     }