文章详情

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

请输入下面的图形验证码

提交验证

短信预约提醒成功

Java8通过CompletableFuture怎么实现异步回调

2023-06-30 11:17

关注

本篇内容介绍了“Java8通过CompletableFuture怎么实现异步回调”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!

1 什么是CompletableFuture?

CompletableFuture是Java 8 中新增的一个类,它是对Future接口的扩展。从下方的类继承关系图中我们看到其不仅实现了Future接口,还有CompletionStage接口,当Future需要显示地完成时,可以使用CompletionStage接口去支持完成时触发的函数和操作,当2个以上线程同时尝试完成、异常完成、取消一个CompletableFuture时,只有一个能成功。

CompletableFuture主要作用就是简化我们异步编程的复杂性,支持函数式编程,可以通过回调的方式处理计算结果。

Java8通过CompletableFuture怎么实现异步回调

2 为什么会有CompletableFuture ?

在java5中,JDK为我们提供了Callable和Future,使我们可以很容易的完成异步任务结果的获取,但是通过Future的get获取异步任务结果会导致主线程的阻塞,这样在某些场景下是非常消耗CPU资源的,进而Java8为我们提供了CompletableFuture,使我们无需阻塞等待,而是通过回调的方式去处理结果,并且还支持流式处理、组合异步任务等操作。

如果不熟悉CallableFuture的,可以看小编之前更新的这篇文章Java从源码看异步任务计算FutureTask

3 CompletableFuture 简单使用

下面我们就CompletableFuture 的使用进行简单分类:

创建任务:

异步回调:

组合处理:

具体内容请参照以下案例:

    public static void main(String[] args) throws Exception {        // 1.带返回值的异步任务(不指定线程池,默认ForkJoinPool.commonPool(),单核ThreadPerTaskExecutor)        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {            return 1 + 1;        });        System.out.println("cf1 result: " + cf1.get());        // 2.无返回值的异步任务(不指定线程池,默认ForkJoinPool.commonPool(),单核ThreadPerTaskExecutor)        CompletableFuture cf2 = CompletableFuture.runAsync(() -> {            int a = 1 + 1;        });        System.out.println("cf2 result: " + cf2.get());        // 3.指定线程池的带返回值的异步任务,runAsync同理        CompletableFuture<Integer> cf3 = CompletableFuture.supplyAsync(() -> {            return 1 + 1;        }, Executors.newCachedThreadPool());        System.out.println("cf3 result: " + cf3.get());        // 4.回调,任务执行完成后执行的动作        CompletableFuture<Integer> cf4 = cf1.thenApply((result) -> {            System.out.println("cf4回调拿到cf1的结果 result : " + result);            return result + 1;        });        System.out.println("cf4 result: " + cf4.get());        // 5.异步回调(将回调任务提交到线程池),任务执行完成后执行的动作后异步执行        CompletableFuture<Integer> cf5 = cf1.thenApplyAsync((result) -> {            System.out.println("cf5回调拿到cf1的结果 result : " + result);            return result + 1;        });        System.out.println("cf5 result: " + cf5.get());        // 6.回调(同thenApply但无返回结果),任务执行完成后执行的动作        CompletableFuture cf6 = cf1.thenAccept((result) -> {            System.out.println("cf6回调拿到cf1的结果 result : " + result);        });        System.out.println("cf6 result: " + cf6.get());        // 7.回调(同thenAccept但无入参),任务执行完成后执行的动作        CompletableFuture cf7 = cf1.thenRun(() -> {        });        System.out.println("cf7 result: " + cf7.get());        // 8.异常回调,任务执行出现异常后执行的动作        CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> {            throw new RuntimeException("出现异常");        });        CompletableFuture<Integer> cf8 = cf.exceptionally((result) -> {            return -1;        });        System.out.println("cf8 result: " + cf8.get());        // 9.当某个任务执行完成后执行的回调方法,会将执行结果或者执行期间抛出的异常传递给回调方法        //   如果是正常执行则异常为null,回调方法对应的CompletableFuture的result和该任务一致;        //   如果该任务正常执行,则get方法返回执行结果,如果是执行异常,则get方法抛出异常。        CompletableFuture<Integer> cf9 = cf1.handle((a, b) -> {            if (b != null) {                b.printStackTrace();            }            return a;        });        System.out.println("cf9 result: " + cf9.get());        // 10 与handle类似,无返回值        try {            CompletableFuture<Integer> cf10 = cf.whenComplete((a, b) -> {                if (b != null) {                    b.printStackTrace();                }            });            System.out.println("cf10 result: " + cf10.get());        } catch (Exception e) {            System.out.println("cf10 出现异常!!!");        }        // 11 组合处理(两个都完成,然后执行)有入参,有返回值        CompletableFuture<Integer> cf11 = cf1.thenCombine(cf3, (r1, r2) -> {            return r1 + r2;        });        System.out.println("cf11 result: " + cf11.get());        // 12 组合处理(两个都完成,然后执行)有入参,无返回值        CompletableFuture cf12 = cf1.thenAcceptBoth(cf3, (r1, r2) -> {        });        System.out.println("cf12 result: " + cf12.get());        // 13 组合处理(两个都完成,然后执行)无入参,无返回值        CompletableFuture cf13 = cf1.runAfterBoth(cf3, () -> {        });        System.out.println("cf13 result: " + cf13.get());        // 14 组合处理(有一个完成,然后执行)有入参,有返回值        CompletableFuture<Integer> cf14 = cf1.applyToEither(cf3, (r) -> {            return r;        });        System.out.println("cf14 result: " + cf14.get());        // 15 组合处理(有一个完成,然后执行)有入参,无返回值        CompletableFuture cf15 = cf1.acceptEither(cf3, (r) -> {        });        System.out.println("cf15 result: " + cf15.get());        // 16 组合处理(有一个完成,然后执行)无入参,无返回值        CompletableFuture cf16 = cf1.runAfterEither(cf3, () -> {        });        System.out.println("cf16 result: " + cf16.get());        // 17 方法执行后返回一个新的CompletableFuture        CompletableFuture<Integer> cf17 = cf1.thenCompose((r) -> {            return CompletableFuture.supplyAsync(() -> {                return 1 + 1;            });        });        System.out.println("cf17 result: " + cf17.get());        // 18 多个任务都执行成功才会继续执行        CompletableFuture.allOf(cf1,cf2,cf3).whenComplete((r, t) -> {            System.out.println(r);        });        // 18 多个任务任意一个执行成功就会继续执行        CompletableFuture.anyOf(cf1,cf2,cf3).whenComplete((r, t) -> {            System.out.println(r);        });    }

4 CompletableFuture 源码分析

首先我们可以从注释中看到,它对CompletionStageFuture接口扩展的一些描述,这些也是它的一些重点。

除了直接操作状态和结果的相关方法外,CompletableFuture还实现了CompletionStage接口的如下策略:

不了解ForkJoinPool的可以阅读小编之前更新的这篇文章一文带你了解Java中的ForkJoin。

CompletableFuture实现了Future接口的如下策略:

4.1 创建异步任务

我们先看一下CompletableFuture是如何创建异步任务的,我们可以看到起创建异步任务的核心实现是两个入参,一个入参是Executor,另一个入参是Supplier(函数式编程接口)。其中也提供了一个入参的重载,一个入参的重载方法会获取默认的Executor,当系统是单核的会使用ThreadPerTaskExecutor,多核时使用ForkJoinPool.commonPool()

注意:这里默认ForkJoinPool.commonPool()线程池,如果所有异步任务都使用该线程池话,出现问题不容易定位,如果长时间占用该线程池可能影响其他业务的正常操作,stream的并行流也是使用的该线程池。

其中还封装了静态内部类AsyncSupply,该类代表这个异步任务,实现了Runnable,重写了run方法。

    private static final Executor asyncPool = useCommonPool ?        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();    private static final boolean useCommonPool =        (ForkJoinPool.getCommonPoolParallelism() > 1);public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {        return asyncSupplyStage(asyncPool, supplier);    }    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;    }    static final class AsyncSupply<T> extends ForkJoinTask<Void>            implements Runnable, AsynchronousCompletionTask {        CompletableFuture<T> dep; Supplier<T> fn;        AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {            this.dep = dep; this.fn = fn;        }        public final Void getRawResult() { return null; }        public final void setRawResult(Void v) {}        public final boolean exec() { run(); return true; }        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类是一个函数式的接口,@FunctionalInterface注解就是函数式编程的标记。

package java.util.function;@FunctionalInterfacepublic interface Supplier<T> {    T get();}

4.2 异步任务回调

异步任务回调,我们以thenApply/thenApplyAsync为例来看一下其实现原理,方法名含有Async的会传入asyncPool。uniApplyStage方法通过判断e是否有值,来区分是从哪个方法进来的。thenApply不会传入 Executor,它优先让当前线程来执行后续 stage 的任务。

thenApplyAsync会传入一个 Executor,它总是让 Executor 线程池里面的线程来执行后续 stage 的任务。

    public <U> CompletableFuture<U> thenApply(        Function<? super T,? extends U> fn) {        return uniApplyStage(null, fn);    }    public <U> CompletableFuture<U> thenApplyAsync(        Function<? super T,? extends U> fn) {        return uniApplyStage(asyncPool, fn);    }    private <V> CompletableFuture<V> uniApplyStage(        Executor e, Function<? super T,? extends V> f) {        if (f == null) throw new NullPointerException();        CompletableFuture<V> d =  new CompletableFuture<V>();        // Async直接进入,不是Async执行uniApply尝试获取结果        if (e != null || !d.uniApply(this, f, null)) {            UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);            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;        // 判断当前CompletableFuture是否已完成,如果没完成则返回false;如果完成了则执行下面的逻辑。        if (a == null || (r = a.result) == null || f == null)            return false;        tryComplete: if (result == null) {            // 判断任务结果是否是AltResult类型            if (r instanceof AltResult) {                if ((x = ((AltResult)r).ex) != null) {                    completeThrowable(x, r);                    break tryComplete;                }                r = null;            }            try {                // 判断当前任务是否可以执行                if (c != null && !c.claim())                    return false;                // 获取任务结果                @SuppressWarnings("unchecked") S s = (S) r;                // 执行                completeValue(f.apply(s));            } catch (Throwable ex) {                completeThrowable(ex);            }        }        return true;    }    static final class UniApply<T,V> extends UniCompletion<T,V> {        Function<? super T,? extends V> fn;        UniApply(Executor executor, CompletableFuture<V> dep,                 CompletableFuture<T> src,                 Function<? super T,? extends V> fn) {            super(executor, dep, src); this.fn = fn;        }        final CompletableFuture<V> tryFire(int mode) {            CompletableFuture<V> d; CompletableFuture<T> a;            if ((d = dep) == null ||                !d.uniApply(a = src, fn, mode > 0 ? null : this))                return null;            dep = null; src = null; fn = null;            return d.postFire(a, mode);        }    }    final void push(UniCompletion<?,?> c) {        if (c != null) {            while (result == null && !tryPushStack(c))                lazySetNext(c, null); // clear on failure        }    }    final boolean completeValue(T t) {        return UNSAFE.compareAndSwapObject(this, RESULT, null,                                           (t == null) ? NIL : t);    }

4.3 异步任务组合

我们再thenCombine方法为例看一下CompletableFuture是如何处理组合任务的,我们可以看到thenCombine的源码与thenApply的源码基本上是一直的,只不过组合的时候不仅仅是判断一个,需要集合具体场景,判断多个CompletableFuture

    public <U,V> CompletableFuture<V> thenCombine(        CompletionStage<? extends U> other,        BiFunction<? super T,? super U,? extends V> fn) {        return biApplyStage(null, other, fn);    }    private <U,V> CompletableFuture<V> biApplyStage(        Executor e, CompletionStage<U> o,        BiFunction<? super T,? super U,? extends V> f) {        CompletableFuture<U> b;        if (f == null || (b = o.toCompletableFuture()) == null)            throw new NullPointerException();        CompletableFuture<V> d = new CompletableFuture<V>();        if (e != null || !d.biApply(this, b, f, null)) {            BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);            bipush(b, c);            c.tryFire(SYNC);        }        return d;    }    final <R,S> boolean biApply(CompletableFuture<R> a,                                CompletableFuture<S> b,                                BiFunction<? super R,? super S,? extends T> f,                                BiApply<R,S,T> c) {        Object r, s; Throwable x;        // 此处不止要判断a还得判断b        if (a == null || (r = a.result) == null ||            b == null || (s = b.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;            }            // 这里不止判断a的结果r还要判断b的结果s            if (s instanceof AltResult) {                if ((x = ((AltResult)s).ex) != null) {                    completeThrowable(x, s);                    break tryComplete;                }                s = null;            }            // 最后将rr, ss传入            try {                if (c != null && !c.claim())                    return false;                @SuppressWarnings("unchecked") R rr = (R) r;                @SuppressWarnings("unchecked") S ss = (S) s;                completeValue(f.apply(rr, ss));            } catch (Throwable ex) {                completeThrowable(ex);            }        }        return true;    }    static final class BiApply<T,U,V> extends BiCompletion<T,U,V> {        BiFunction<? super T,? super U,? extends V> fn;        BiApply(Executor executor, CompletableFuture<V> dep,                CompletableFuture<T> src, CompletableFuture<U> snd,                BiFunction<? super T,? super U,? extends V> fn) {            super(executor, dep, src, snd); this.fn = fn;        }        // tryFire方法也同样的多可个b        final CompletableFuture<V> tryFire(int mode) {            CompletableFuture<V> d;            CompletableFuture<T> a;            CompletableFuture<U> b;            if ((d = dep) == null ||                !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this))                return null;            dep = null; src = null; snd = null; fn = null;            return d.postFire(a, b, mode);        }    }

“Java8通过CompletableFuture怎么实现异步回调”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注编程网网站,小编将为大家输出更多高质量的实用文章!

阅读原文内容投诉

免责声明:

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

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

软考中级精品资料免费领

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

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

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

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

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

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

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