CompletableFuture 原理分析

分手后的思念是犯贱 2024-03-24 09:11 42阅读 0赞

上篇主要介绍了CompletableFuture的特性以及一些使用方法，本章节我们主要来聊一聊CompletableFuture的回调功能以及异步工作原理的实现

### 代码结构 ###

#### CompletableFuture 代码结构 ####

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
        volatile Object result;       // Either the result or boxed AltResult
        volatile Completion stack;    // Top of Treiber stack of dependent actions
       
    }

我们主要看下 result和 stack 2个属性

*  **result**：存储CompletableFuture的返回，或者存储异常对象AltResult。
 *  **stack**：是 CompletableFuture.Completion对象，表示操作数栈栈顶，在进行CompletableFuture链式调用的过程中，所有链式调用的CompletableFuture任务都会被压入该stack中，在任务调用的过程按**后进先出**的顺序出栈执行完所有任务

#### Completion 代码结构 ####

abstract static class Completion extends ForkJoinTask<Void>
        implements Runnable, AsynchronousCompletionTask {
        volatile Completion next;      // Treiber stack link
        ...
    }

**next**：存储下一个任务链式调用栈

#### UniCompletion的代码结构 ####

abstract static class UniCompletion<T,V> extends Completion {
            Executor executor;                 // executor to use (null if none)
            CompletableFuture<V> dep;          // the dependent to complete
            CompletableFuture<T> src;          // source for action
     
            UniCompletion(Executor executor, CompletableFuture<V> dep,
                          CompletableFuture<T> src) {
                this.executor = executor; this.dep = dep; this.src = src;
            }
    }

UniCompletion继承Completion类，包含以下几个参数：

*  **executor**：异步任务执行器，如果为空则有主线程执行任务不进行异步执行。
 *  **dep**：指向当前任务构建的CompletabueFuture
 *  **src**：指向源CompletableFuture任务

#### 回调原理 ####

示例

public static void main(String[] args) {
            CompletableFuture<String> demoFuture= CompletableFuture.completedFuture("Demo Future");
            log.info(demoFuture.thenApply((r) -> r + " Then Apply").join());
            demoFuture.thenAccept((r) -> log.info(r)).thenAccept((Void) -> log.info("Void"));
        }
    
    
    控制台输出
    
    Demo Future Then Apply
    Demo Future
    Void

#### 代码流程分析 ####

我们通过 【 CompletableFuture.completedFuture(“Demo Future”)】 这段代码创建了一个 CompletableFuture 对象当我们执行demoFuture.thenApply 方法时会引发一个入栈操作，demo`Future`引用的`CompletableFuture`的`stack`属性将会指向demo`Future``.thenApply()`结果返回的新`CompletableFuture`对象，而新`CompletableFuture`对象的`src`属性将指向来源`CompletableFuture`即demo`Future`所引用的对象。

这段话看起来比较绕口，我们画个简单的示意图来表示下

![40e0081aaa28da1b07fa6b1d1b540f76.png][]

在执行完demoFuture`.thenAccept()`的时候，`thenAccept`返回的任务将被压入栈顶，`next`指向上一个代码段的返回对象，在`thenAccept`返回的新`CompletableFuture`对象中在进行一次`thenAccept`的调用，就再产生一个新的`CompletableFuture`对象，`dep`属性就指向最新的`CompletableFuture`对象

![af16d2d547072957a5643f8fd492718c.png][]

#### thenApply 分析 ####

public <U> CompletableFuture<U> thenApply(
        Function<? super T,? extends U> fn) {
        return uniApplyStage(null, fn);
    }
     
    private <V> CompletableFuture<V> uniApplyStage(
        Executor e, Function<? super T,? extends V> f) {
        if (f == null) throw new NullPointerException();
        // 创建一个新的CompletableFuture对象
        CompletableFuture<V> d =  new CompletableFuture<V>();
        // e：如果是异步调用直接执行代码块
        // !d.uniApply：执行任务，如果返回false则任务未执行需入栈
        if (e != null || !d.uniApply(this, f, null)) {
            UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
            // 创建出新的UniApply对象进行入栈
            push(c);
            // 尝试执行任务
            c.tryFire(SYNC);
        }
        return d;
    }
     
    final <S> boolean uniApply(CompletableFuture<S> a,
                               Function<? super S,? extends T> f,
                               UniApply<S,T> c) {
        Object r; Throwable x;
        // 任务未完成结果为null直接返回false
        if (a == null || (r = a.result) == null || f == null)
            return false;
        // 验证是否出现异常结果，如有则任务执行结束
        tryComplete: if (result == null) {
            if (r instanceof AltResult) {
                if ((x = ((AltResult)r).ex) != null) {
                    completeThrowable(x, r);
                    break tryComplete;
                }
                r = null;
            }
            try {
                // 异步执行任务
                if (c != null && !c.claim())
                    // 任务未执行返回false
                    return false;
                @SuppressWarnings("unchecked") S s = (S) r;
                // 任务执行完成将结果写入result
                completeValue(f.apply(s));
            } catch (Throwable ex) {
                completeThrowable(ex);
            }
        }
        return true;
    }

以上代码片段主要描述了`CompletableFuture`在执行任务时会创建出新的`CompletableFuture`对象，使用新对象执行任务并获取结果使用`CAS`写入到`result`属性，如果任务未执行将压入栈顶，再重新尝试任务执行，在`CompletableFuture`其他方法的调用也都大同小异，这里不在逐一分析，可自行打开源码阅读便于理解

#### 异步加载原理 ####

ublic 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>();
        // 如果是异步任务，这里的参数e不会为空，也就是会将任务执行压入栈顶
        if (e != null || !d.uniAccept(this, f, null)) {
            UniAccept<T> c = new UniAccept<T>(e, d, this, f);
            push(c);
            // 重点还是这个入口
            c.tryFire(SYNC);
        }
        return d;
    }
     
    static final class UniAccept<T> extends UniCompletion<T,Void> {
        Consumer<? super T> fn;
        UniAccept(Executor executor, CompletableFuture<Void> dep,
                  CompletableFuture<T> src, Consumer<? super T> fn) {
            super(executor, dep, src); this.fn = fn;
        }
        final CompletableFuture<Void> tryFire(int mode) {
            CompletableFuture<Void> d; CompletableFuture<T> a;
            // dep为空即任务已被执行过，直接返回null
            // uniAccept()结果为false，可能是任务执行中未完成，也可能是由线程池中的其他任务执行完成
            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);
        }
    }
     
    final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
        if (a != null && a.stack != null) {
            // postComplete调用过来的，或者上一个任务执行完成，清空栈数据，否则调用postComplete完成任务
            if (mode < 0 || a.result == null)
                a.cleanStack();
            else
                // 完成任务执行并进行出栈
                a.postComplete();
        }
        if (result != null && stack != null) {
            if (mode < 0)
                // postComplete调用过来的任务已完成
                return this;
            else
                // 完成任务执行并进行出栈
                postComplete();
        }
        return null;
    }

`CompletableFuture`进行异步主要是通过将任务压入栈顶后`tryFire`方法进行异步处理，如果任务未被执行则会通过`postFire`方法有线程池中的线程进行任务执行，任务执行结果再使用`CAS`将结果返回到`result`，其他线程即可得知任务是否被执行过，如果当前现场判断当前任务为被执行，则调用`postComplete()`执行完成任务。

通过对CompletableFuture构造参数的解读，我们发现里面有个核心参数—-线程池（负责异步调用任务），如果想要使用好异步加载，必须对线程池知识有一定的掌握，线程池可以参考之前的文章  
  
[ThreadPoolExecutor原理解析（一）][ThreadPoolExecutor]  
[@Async为什么要使用自定义线程池][Async]

--------------------

[参考][Link 1]

[40e0081aaa28da1b07fa6b1d1b540f76.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/468170104f2842ff900210fa74dc5e21.png
[af16d2d547072957a5643f8fd492718c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/a1992f4f830d41428bf1c4ce88aa8ca4.png
[ThreadPoolExecutor]: https://mp.csdn.net/mp_blog/creation/editor/new/129161821
[Async]: https://mp.csdn.net/mp_blog/creation/editor/new/129731730
[Link 1]: https://tech.tuhu.cn/?p=7448