Java8中CompletableFuture使用场景与实现原理

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

上面的结论是通过阅读源代码得到的。下面我们深入源代码。

3.源码追踪

创建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的编码，真的是艺术。代码的复用性全体现在逻辑上了。

到此这篇关于java8中completablefuture使用场景与实现原理的文章就介绍到这了,更多相关completablefuture使用场景与原理内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！