文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Java8中CompletableFuture怎么使用

2023-06-29 02:22

关注

今天小编给大家分享一下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。

可以实现如下等功能

消费和运行的区别:

消费使用执行结果。运行则只是运行特定任务。具体其他功能大家可以根据需求自行查看。

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线程调用。

这里说明一下,如果是同一任务的依赖任务有多个:

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

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

该方法会调用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方法是逻辑:

this的run任务如下。也就是在异步线程同步调用tryFire方法。达到其被异步线程执行的目的。

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

看完上面的逻辑,我们基本理解依赖任务的逻辑。

其实就是先判断源任务是否完成,如果完成,直接在对应线程执行以来任务(如果是同步,则在当前线程处理,否则在异步线程处理)

如果任务没有完成,直接返回,因为等任务完成之后会通过postComplete去触发调用依赖任务。

postComplete方法

final void postComplete() {        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,则不调用。

堆栈的内容其实就是在依赖任务创建的时候加入进去的。上面我们已经提到过。

以上就是“Java8中CompletableFuture怎么使用”这篇文章的所有内容,感谢各位的阅读!相信大家阅读完这篇文章都有很大的收获,小编每天都会为大家更新不同的知识,如果还想学习更多的知识,请关注编程网行业资讯频道。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯