1.概述

CompletableFuture是jdk1.8引入的实现类。扩大了Future和CompletionStage,是一个能够在工作实现阶段触发一些操作Future。简略的来讲就是能够实现异步回调。

2.为什么引入CompletableFuture

对于jdk1.5的Future,尽管提供了异步解决工作的能力,然而获取后果的形式很不优雅,还是须要通过阻塞(或者轮训)的形式。如何防止阻塞呢?其实就是注册回调。

业界联合观察者模式实现异步回调。也就是当工作执行实现后去告诉观察者。比方Netty的ChannelFuture,能够通过注册监听实现异步后果的解决。

Netty的ChannelFuture
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;  }  

通过addListener办法注册监听。如果工作实现,会调用notifyListeners告诉。

CompletableFuture通过扩大Future,引入函数式编程,通过回调的形式去处理结果。

3.性能

CompletableFuture的性能次要体现在他的CompletionStage。

能够实现如下等性能

  • 转换(thenCompose)
  • 组合(thenCombine)
  • 生产(thenAccept)
  • 运行(thenRun)。
  • 带返回的生产(thenApply)

生产和运行的区别:

生产应用执行后果。运行则只是运行特定工作。具体其余性能大家能够依据需要自行查看。

CompletableFuture借助CompletionStage的办法能够实现链式调用。并且能够抉择同步或者异步两种形式。

这里举个简略的例子来体验一下他的性能。

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  

依据后果咱们能够看到会有序执行对应工作。

留神:

如果是同步执行cf.thenRun。他的执行线程可能main线程,也可能是执行源工作的线程。如果执行源工作的线程在main调用之前执行完了工作。那么cf.thenRun办法会由main线程调用。

这里阐明一下,如果是同一工作的依赖工作有多个:

  • 如果这些依赖工作都是同步执行。那么如果这些工作被以后调用线程(main)执行,则是有序执行,如果被执行源工作的线程执行,那么会是倒序执行。因为外部工作数据结构为LIFO。
  • 如果这些依赖工作都是异步执行,那么他会通过异步线程池去执行工作。不能保障工作的执行程序。

下面的论断是通过浏览源代码失去的。上面咱们深刻源代码。

4.源码追踪

创立CompletableFuture

创立的办法有很多,甚至能够间接new一个。咱们来看一下supplyAsync异步创立的办法。

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;  }  

入参Supplier,带返回值的函数。如果是异步办法,并且传递了执行器,那么会应用传入的执行器去执行工作。否则采纳公共的ForkJoin并行线程池,如果不反对并行,新建一个线程去执行。

这里咱们须要留神ForkJoin是通过守护线程去执行工作的。所以必须有非守护线程的存在才行。

asyncSupplyStage办法
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;  }  

这里会创立一个用于返回的CompletableFuture。

而后结构一个AsyncSupply,并将创立的CompletableFuture作为结构参数传入。

那么,工作的执行齐全依赖AsyncSupply。

AsyncSupply#run
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();      }  }  
  1. 该办法会调用Supplier的get办法。并将后果设置到CompletableFuture中。咱们应该分明这些操作都是在异步线程中调用的。
  2. d.postComplete办法就是告诉工作执行实现。触发后续依赖工作的执行,也就是实现CompletionStage的关键点。

在看postComplete办法之前咱们先来看一下创立依赖工作的逻辑。

thenAcceptAsync办法
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;  }  

下面提到过。thenAcceptAsync是用来生产CompletableFuture的。该办法调用uniAcceptStage。

uniAcceptStage逻辑:

  1. 结构一个CompletableFuture,次要是为了链式调用。
  2. 如果为异步工作,间接返回。因为源工作完结后会触发异步线程执行对应逻辑。
  3. 如果为同步工作(e==null),会调用d.uniAccept办法。这个办法在这里逻辑:如果源工作实现,调用f,返回true。否则进入if代码块(Mark 1)。
  4. 如果是异步工作间接进入if(Mark 1)。

Mark1逻辑:

  1. 结构一个UniAccept,将其push入栈。这里通过CAS实现乐观锁实现。
  2. 调用c.tryFire办法。
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);  }  
  1. 会调用d.uniAccept办法。其实该办法判断源工作是否实现,如果实现则执行依赖工作,否则返回false。
  2. 如果依赖工作曾经执行,调用d.postFire,次要就是Fire的后续解决。依据不同模式逻辑不同。

这里简略说一下,其实mode有同步异步,和迭代。迭代为了防止有限递归。

这里强调一下d.uniAccept办法的第三个参数。

如果是异步调用(mode>0),传入null。否则传入this。

区别看上面代码。c不为null会调用c.claim办法。

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;  }  

claim办法是逻辑:

  • 如果异步线程为null。阐明同步,那么间接返回true。最初下层函数会调用f.accept(s)同步执行工作。
  • 如果异步线程不为null,那么应用异步线程去执行this。

this的run工作如下。也就是在异步线程同步调用tryFire办法。达到其被异步线程执行的目标。

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

看完下面的逻辑,咱们根本了解依赖工作的逻辑。

其实就是先判断源工作是否实现,如果实现,间接在对应线程执行以来工作(如果是同步,则在以后线程解决,否则在异步线程解决)

如果工作没有实现,间接返回,因为等工作实现之后会通过postComplete去触发调用依赖工作。

postComplete办法
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;          }      }  }  

在源工作实现之后会调用。

其实逻辑很简略,就是迭代堆栈的依赖工作。调用h.tryFire办法。NESTED就是为了防止递归死循环。因为FirePost会调用postComplete。如果是NESTED,则不调用。

堆栈的内容其实就是在依赖工作创立的时候退出进去的。下面咱们曾经提到过。

4.总结

根本上述源码曾经剖析了逻辑。

因为波及异步等操作,咱们须要理一下(这里针对全异步工作):

  1. 创立CompletableFuture胜利之后会通过异步线程去执行对应工作。
  2. 如果CompletableFuture还有依赖工作(异步),会将工作退出到CompletableFuture的堆栈保存起来。以供后续实现后执行依赖工作。
当然,创立依赖工作并不只是将其退出堆栈。如果源工作在创立依赖工作的时候曾经执行实现,那么以后线程会触发依赖工作的异步线程间接解决依赖工作。并且会通知堆栈其余的依赖工作源工作曾经实现。

次要是思考代码的复用。所以逻辑绝对难了解。

postComplete办法会被源工作线程执行完源工作后调用。同样也可能被依赖工作线程后调用。

执行依赖工作的办法次要就是靠tryFire办法。因为这个办法可能会被多种不同类型线程触发,所以逻辑也绕一点。(其余依赖工作线程、源工作线程、以后依赖工作线程)

  • 如果是以后依赖工作线程,那么会执行依赖工作,并且会告诉其余依赖工作。
  • 如果是源工作线程,和其余依赖工作线程,则将工作转换给依赖线程去执行。不须要告诉其余依赖工作,防止死递归。

不得不说Doug Lea的编码,真的是艺术。代码的复用性整体当初逻辑上了。

起源:blog.csdn.net/weixin_39332800/article/details/108185931

近期热文举荐:

1.1,000+ 道 Java面试题及答案整顿(2022最新版)

2.劲爆!Java 协程要来了。。。

3.Spring Boot 2.x 教程,太全了!

4.别再写满屏的爆爆爆炸类了,试试装璜器模式,这才是优雅的形式!!

5.《Java开发手册(嵩山版)》最新公布,速速下载!

感觉不错,别忘了顺手点赞+转发哦!