一.使用

  1. public class Test {
  3.     public static void main(String[] args) throws InterruptedException {
  4.         ExecutorService executorService = Executors.newFixedThreadPool(2);
  6.         CompletableFuture.supplyAsync(() -> {
  7.             try {
  8.                 Thread.sleep(2000);
  9.             } catch (Exception e) {
  10.                 e.printStackTrace();
  11.             }
  12.             System.out.println("AAA " + Thread.currentThread().getName());
  13.             return "hello ";
  14.         }, executorService).thenAccept(s -> {
  15.             System.out.println(s + "world");
  16.             System.out.println("BBB " + Thread.currentThread().getName());
  17.         });
  19.         System.out.println(Thread.currentThread().getName());
  20.         Thread.sleep(40000);
  21.     }
  22. }
  24. 输出：
  25. main
  26. AAA pool-1-thread-1
  27. hello world
  28. BBB pool-1-thread-1

二.源码

  1. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
  2.                                                    Executor executor) {
  3.     return asyncSupplyStage(screenExecutor(executor), supplier);
  4. }
  6. static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
  7.                                                  Supplier<U> f) {
  8.     if (f == null) throw new NullPointerException();
  9.     CompletableFuture<U> d = new CompletableFuture<U>();
  10.     e.execute(new AsyncSupply<U>(d, f));
  11.     return d;
  12. }
  14. static final class AsyncSupply<T> extends ForkJoinTask<Void>
  15.         implements Runnable, AsynchronousCompletionTask {
  16.     CompletableFuture<T> dep; Supplier<T> fn;
  17.     AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
  18.         this.dep = dep; this.fn = fn;
  19.     }
  21.     public void run() {
  22.         CompletableFuture<T> d; Supplier<T> f;
  23.         if ((d = dep) != null && (f = fn) != null) {
  24.             dep = null; fn = null;
  25.             if (d.result == null) {
  26.                 try {
  27.                     // 执行到这Thread.sleep(2000);
  28.                     d.completeValue(f.get());
  29.                 } catch (Throwable ex) {
  30.                     d.completeThrowable(ex);
  31.                 }
  32.             }
  33.             d.postComplete();
  34.         }
  35.     }
  36. }
  38. public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
  39.     return uniAcceptStage(null, action);
  40. }
  42. private CompletableFuture<Void> uniAcceptStage(Executor e,
  43.                                                Consumer<? super T> f) {
  44.     if (f == null) throw new NullPointerException();
  45.     CompletableFuture<Void> d = new CompletableFuture<Void>();
  46.     if (e != null || !d.uniAccept(this, f, null)) {
  47.         UniAccept<T> c = new UniAccept<T>(e, d, this, f);
  48.         // 入栈volatile Completion stack;
  49.         push(c);
  50.         c.tryFire(SYNC);
  51.     }
  52.     return d;
  53. }
  55. final <S> boolean uniAccept(CompletableFuture<S> a,
  56.                             Consumer<? super S> f, UniAccept<S> c) {
  57.     Object r; Throwable x;
  58.     // a.result == null
  59.     if (a == null || (r = a.result) == null || f == null)
  60.         return false;
  61. }
  63. static final class UniAccept<T> extends UniCompletion<T,Void> {
  64.     Consumer<? super T> fn;
  65.     UniAccept(Executor executor, CompletableFuture<Void> dep,
  66.               CompletableFuture<T> src, Consumer<? super T> fn) {
  67.         super(executor, dep, src); this.fn = fn;
  68.     }
  69.     final CompletableFuture<Void> tryFire(int mode) {
  70.         CompletableFuture<Void> d; CompletableFuture<T> a;
  71.         if ((d = dep) == null ||
  72.             !d.uniAccept(a = src, fn, mode > 0 ? null : this))
  73.             // 执行到这
  74.             return null;
  75.         dep = null; src = null; fn = null;
  76.         return d.postFire(a, mode);
  77.     }
  78. }
  80. final void postComplete() {
  81.     CompletableFuture<?> f = this; Completion h;
  82.     // f.stack != null
  83.     while ((h = f.stack) != null ||
  84.            (f != this && (h = (f = this).stack) != null)) {
  85.         CompletableFuture<?> d; Completion t;
  86.         // stack = h.next
  87.         if (f.casStack(h, t = h.next)) {
  88.             if (t != null) {
  89.                 if (f != this) {
  90.                     pushStack(h);
  91.                     continue;
  92.                 }
  93.                 h.next = null;    // detach
  94.             }
  95.             f = (d = h.tryFire(NESTED)) == null ? this : d;
  96.         }
  97.     }
  98. }
  100. final CompletableFuture<Void> tryFire(int mode) {
  101.     CompletableFuture<Void> d; CompletableFuture<T> a;
  102.     if ((d = dep) == null ||
  103.         !d.uniAccept(a = src, fn, mode > 0 ? null : this))
  104.         return null;
  105.     dep = null; src = null; fn = null;
  106.     return d.postFire(a, mode);
  107. }
  110. final <S> boolean uniAccept(CompletableFuture<S> a,
  111.                         Consumer<? super S> f, UniAccept<S> c) {
  112.     Object r; Throwable x;
  113.     // a.result:hello
  114.     if (a == null || (r = a.result) == null || f == null)
  115.         return false;
  116.     tryComplete: if (result == null) {
  117.         if (r instanceof AltResult) {
  118.             if ((x = ((AltResult)r).ex) != null) {
  119.                 completeThrowable(x, r);
  120.                 break tryComplete;
  121.             }
  122.             r = null;
  123.         }
  124.         try {
  125.             if (c != null && !c.claim())
  126.                 return false;
  127.             @SuppressWarnings("unchecked") S s = (S) r;
  128.             // s:hello
  129.             f.accept(s);
  130.             completeNull();
  131.         } catch (Throwable ex) {
  132.             completeThrowable(ex);
  133.         }
  134.     }
  135.     return true;
  136. }