CompletableFuture使用场景和实现原理

public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {      checkNotNull(listener, "listener");      synchronized (this) {          addListener0(listener);      }      if (isDone()) {          notifyListeners();      }      return this;  }  private boolean setValue0(Object objResult) {      if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||          RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {          if (checkNotifyWaiters()) {              notifyListeners();          }          return true;      }      return false;  }

public static void thenApply() {      ExecutorService executorService = Executors.newFixedThreadPool(2);      CompletableFuture cf = CompletableFuture.supplyAsync(() -> {          try {              //  Thread.sleep(2000);          } catch (Exception e) {              e.printStackTrace();          }          System.out.println("supplyAsync " + Thread.currentThread().getName());          return "hello";      }, executorService).thenApplyAsync(s -> {          System.out.println(s + "world");          return "hhh";      }, executorService);      cf.thenRunAsync(() -> {          System.out.println("ddddd");      });      cf.thenRun(() -> {          System.out.println("ddddsd");      });      cf.thenRun(() -> {          System.out.println(Thread.currentThread());          System.out.println("dddaewdd");      });  }

supplyAsync pool-1-thread-1  helloworld  ddddd  ddddsd  Thread[main,5,main]  dddaewdd  

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,                                                     Executor executor) {      return asyncSupplyStage(screenExecutor(executor), supplier);  }  static Executor screenExecutor(Executor e) {      if (!useCommonPool && e == ForkJoinPool.commonPool())          return asyncPool;      if (e == null) throw new NullPointerException();      return e;  }

static <U> CompletableFuture<U> asyncSupplyStage(Executor e,                                                   Supplier<U> f) {      if (f == null) throw new NullPointerException();      CompletableFuture<U> d = new CompletableFuture<U>();      e.execute(new AsyncSupply<U>(d, f));      return d;  }

public void run() {      CompletableFuture<T> d; Supplier<T> f;      if ((d = dep) != null && (f = fn) != null) {          dep = null; fn = null;          if (d.result == null) {              try {                  d.completeValue(f.get());              } catch (Throwable ex) {                  d.completeThrowable(ex);              }          }          d.postComplete();      }  }

public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {      return uniAcceptStage(asyncPool, action);  }  private CompletableFuture<Void> uniAcceptStage(Executor e,                                                 Consumer<? super T> f) {      if (f == null) throw new NullPointerException();      CompletableFuture<Void> d = new CompletableFuture<Void>();      if (e != null || !d.uniAccept(this, f, null)) {          # 1          UniAccept<T> c = new UniAccept<T>(e, d, this, f);          push(c);          c.tryFire(SYNC);      }      return d;  }

final CompletableFuture<Void> tryFire(int mode) {      CompletableFuture<Void> d; CompletableFuture<T> a;      if ((d = dep) == null ||          !d.uniAccept(a = src, fn, mode > 0 ? null : this))          return null;      dep = null; src = null; fn = null;      return d.postFire(a, mode);  }

try {      if (c != null && !c.claim())          return false;      @SuppressWarnings("unchecked") S s = (S) r;      f.accept(s);      completeNull();  } catch (Throwable ex) {      completeThrowable(ex);  }    final boolean claim() {      Executor e = executor;      if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {          if (e == null)              return true;          executor = null; // disable          e.execute(this);      }      return false;  }

public final void run(){      tryFire(ASYNC);   }

final void postComplete() {      /*       * On each step, variable f holds current dependents to pop       * and run.  It is extended along only one path at a time,       * pushing others to avoid unbounded recursion.       */      CompletableFuture<?> f = this; Completion h;      while ((h = f.stack) != null ||             (f != this && (h = (f = this).stack) != null)) {          CompletableFuture<?> d; Completion t;          if (f.casStack(h, t = h.next)) {              if (t != null) {                  if (f != this) {                      pushStack(h);                      continue;                  }                  h.next = null;    // detach              }              f = (d = h.tryFire(NESTED)) == null ? this : d;          }      }  }