关于java:京东一面说说-CompletableFuture-的实现原理和使用场景我懵了

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

对于 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，引入函数式编程，通过回调的形式去处理结果。

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。
如果这些依赖工作都是异步执行，那么他会通过异步线程池去执行工作。不能保障工作的执行程序。

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

创立 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();}  
}

该办法会调用 Supplier 的 get 办法。并将后果设置到 CompletableFuture 中。咱们应该分明这些操作都是在异步线程中调用的。
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 逻辑：

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

Mark1 逻辑：

结构一个 UniAccept，将其 push 入栈。这里通过 CAS 实现乐观锁实现。
调用 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);  
}

会调用 d.uniAccept 办法。其实该办法判断源工作是否实现，如果实现则执行依赖工作，否则返回 false。
如果依赖工作曾经执行，调用 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，则不调用。

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

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

因为波及异步等操作，咱们须要理一下（这里针对全异步工作）：

创立 CompletableFuture 胜利之后会通过异步线程去执行对应工作。
如果 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 开发手册（嵩山版）》最新公布，速速下载！

感觉不错，别忘了顺手点赞 + 转发哦！

关于java:京东一面说说-CompletableFuture-的实现原理和使用场景我懵了

1. 概述

2. 为什么引入 CompletableFuture

Netty 的 ChannelFuture

3. 性能

4. 源码追踪

创立 CompletableFuture

asyncSupplyStage 办法

AsyncSupply#run

thenAcceptAsync 办法

postComplete 办法

4. 总结