Java异步实现的N种方式

2年前 (2022) 程序员胖胖胖虎阿
353 0 0

背景

异步编程现在受到了越来越多的关注,尤其是在 IO 密集型的业务场景中,相比传统的同步开发模式,异步编程的优势越来越明显,本文介绍Java常见的实现方式;

Future

描述

java.util.concurrent.Future是JDK5引入的,用来获取一个异步计算的结果。可以使用isDone方法检查计算是否完成,也可以使用get阻塞住调用线程,直到计算完成返回结果,使用cancel方法停止任务的执行。

Java异步实现的N种方式

Java异步实现的N种方式

FutureTask.java是对Futre和Runnable最简单的实现,实现了run函数,所以可以直接执行,任务执行结束通过set()保存结果,setException()保存异常信息。通常配合executorService.submit()一起使用,ExecutorService中将任务包装成FutureTask执行execute();

Java异步实现的N种方式

Java异步实现的N种方式

样例

Java异步实现的N种方式

		@Test
    public void futureCallBackTest() throws InterruptedException, ExecutionException {
        System.out.println(printThread("小明点餐"));
        Future<String> future = executorService.submit(() -> {
            System.out.println(printThread("厨师开始炒菜"));
            Thread.sleep(2000);
            System.out.println(printThread( "厨师炒好菜"));
            return "饭菜好了";
        });

        String result = future.get();
        executorService.shutdown();
        System.out.println(printThread(result + ",小明开始吃饭"));
    }

运行结果

Java异步实现的N种方式

优缺点

  • 能获得异步线程执行结果

  • 无法方便得知任务何时完成

  • 在主线程获得任务结果会导致主线程阻塞

  • 复杂一点的情况下,比如多个异步任务的场景,一个异步任务依赖上一个异步任务的执行结果,异步任务合并等,Future无法满足需求

Java异步实现的N种方式

ListenableFuture

描述

Google并发包下的listenableFuture对Java原生的future做了扩展,顾名思义就是使用监听器模式实现的回调,所以叫可监听的future,通过addListener(Runnable listener, Executor executor)方法添加回调任务。

要使用listenableFuture还要结合MoreExecutor线程池,MoreExecutor是对Java原生线程池的封装,比如常用的MoreExecutors.listeningDecorator(threadPool); 修改Java原生线程池的submit方法,封装了future返回listenableFuture。

Java异步实现的N种方式

Java异步实现的N种方式

样例

Java异步实现的N种方式

@Test
    public void listenableFutureTest() throws InterruptedException, ExecutionException {
        System.out.println(printThread("小明点餐"));
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
        ListenableFuture<String> listenableFuture = listeningExecutorService.submit(() -> {
            System.out.println(printThread("厨师开始炒菜"));
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(printThread( "厨师炒好菜"));
            return "饭菜好了";
        });

        Futures.addCallback(listenableFuture, new FutureCallback<String>() {
            @Override
            public void onSuccess(@Nullable String s) {
                System.out.println(printThread(s + ",小明开始吃饭"));
            }

            @Override
            public void onFailure(Throwable throwable) {
                System.out.println(printThread( throwable.getMessage()));
            }
        }, executorService);

        System.out.println(printThread( "小明开始玩游戏"));
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(printThread("小明结束玩游戏"));


        listenableFuture.get();
        listeningExecutorService.shutdown();
        executorService.shutdown();
    }

运行结果

Java异步实现的N种方式

这里的运行结果:小明玩游戏和小明吃饭放在了2个线程,没有阻塞等待。

优缺点

充分利用线程的时间片

回调机制的最大问题是:Callback Hell(回调地狱)

Java异步实现的N种方式

CallbackHell

描述

大量使用 Callback 机制,使应该是先后的业务逻辑在代码形式上表现为层层嵌套,这会导致代码难以理解和维护

样例

Java异步实现的N种方式

@Test
    public void listenableFutureCallbackHellTest() throws InterruptedException, ExecutionException {
        System.out.println(printThread("小明点餐"));
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
        ListenableFuture<String> listenableFuture = listeningExecutorService.submit(() -> {
            System.out.println(printThread("厨师开始做菜"));
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "菜已装盘";
        });
        Futures.addCallback(listenableFuture, new FutureCallback<String>() {
            @Override
            public void onSuccess(@Nullable String s) {
                System.out.println(printThread(s + ",小明开始吃饭"));
                System.out.println(printThread( "小明点了个饮料"));
                ListenableFuture<String> listenableFuture1 = listeningExecutorService.submit(() -> {
                    System.out.println(printThread("服务员拿饮料"));
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return "饮料好了";
                });
                Futures.addCallback(listenableFuture1, new FutureCallback<String>() {
                    @Override
                    public void onSuccess(@Nullable String s) {
                        System.out.println(printThread(s + ",小明开始喝饮料"));
                    }

                    @Override
                    public void onFailure(Throwable throwable) {

                    }
                }, executorService);
            }

            @Override
            public void onFailure(Throwable throwable) {
                System.out.println(printThread( throwable.getMessage()));
            }
        }, executorService);

        System.out.println(printThread( "小明开始玩游戏"));
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(printThread("小明结束玩游戏"));

        listenableFuture.get();
        listeningExecutorService.shutdown();
        executorService.awaitTermination(10, TimeUnit.SECONDS);
        executorService.shutdown();
    }

Java异步实现的N种方式

CompleteableFuture

描述

Java8新增的CompletableFuture类借鉴了Google Guava的ListenableFuture,它包含50多个方法,默认使用forkJoinPool线程池,提供了非常强大的Future扩展功能,可以帮助我们简化异步编程的复杂性,结合函数式编程,通过回调的方式处理计算结果,并且提供了转换和组合CompletableFuture的多种方法,可以满足大部分异步回调场景。

CompletableFuture可以用来以声明式语义构建创建异步任务的编排模式,它可以用于通过声明表示:

  • 将要执行一个异步任务;

  • 将要执行一个异步任务,它必须在一个前驱异步任务完成之后执行,其以前驱任务的输出作为自身的输入;

  • 将要执行一个异步任务,它必须在若干前驱异步任务中的(任意或全部)完成之后执行,其以全部(或任一)前驱任务的输出作为自身的输入;

样例

@Test
    public void completeableFutureTest()  {
        System.out.println(printThread("小明点餐"));

        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(printThread("厨师开始做菜"));
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(printThread("厨师菜做好了"));
            return "菜已装盘";
        });

        CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
            System.out.println(printThread( "小明开始玩游戏"));
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(printThread("小明结束玩游戏"));
        });

        CompletableFuture<Void> completableFuture2 = completableFuture
                .thenAcceptBoth(completableFuture1,(a, b) -> System.out.println(printThread( a + ", 小明开始吃饭,并点了饮料")))
                .thenApplyAsync((b) -> {
                    System.out.println(printThread("服务员拿饮料"));
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return "饮料好了";
                },executorService)
                .thenAcceptAsync((s) -> System.out.println(printThread(s + ",小明开始喝饮料")));

        completableFuture2.join();
    }

Java异步实现的N种方式

方法介绍

创建对象

Java异步实现的N种方式

以Async结尾并且没有指定Executor的方法会使用ForkJoinPool.commonPool()作为它的线程池执行异步代码。

runAsync方法也好理解,它以Runnable函数式接口类型为参数,所以CompletableFuture的计算结果为空。

supplyAsync方法以Supplier<U>函数式接口类型为参数,CompletableFuture的计算结果类型为U。

计算结果完成时的处理

Java异步实现的N种方式

当CompletableFuture的计算结果完成,或者抛出异常的时候,我们可以执行特定的Action;

不以Async结尾的方法由原来的线程计算,以Async结尾的方法由默认的线程池ForkJoinPool.commonPool()或者指定的线程池executor运行;

exceptionally方法返回一个新的CompletableFuture,当原始的CompletableFuture抛出异常的时候,就会触发这个CompletableFuture的计算,调用function计算值;

转换、消费

Java异步实现的N种方式

一个传Function将CompletableFuture中的值转换成另一个值,一个传Consumer将CompletableFuture值消费;

组合

Java异步实现的N种方式

thenCombine用来复合另外一个CompletionStage的结果,两个CompletionStage是并行执行的,它们之间并没有先后依赖顺序;

thenAcceptBoth和runAfterBoth是当两个CompletableFuture都计算完成,acceptEither和applyToEither方法是当任意一个CompletionStage完成的时候执行后续任务;

辅助方法 allOf 和 anyOf

Java异步实现的N种方式

allOf方法是当所有的CompletableFuture都执行完后执行计算;

anyOf方法是当任意一个CompletableFuture执行完后就会执行计算;

Reactor

描述

Reactor 框架是 Pivotal 公司( Spring 家族公司)开发的,实现了 Reactive Programming 思想,符合 Reactive Streams 规范的一项技术;

Java异步实现的N种方式

Reactive

反应式宣言-反应式宣言

Java异步实现的N种方式

Reactive Streams

介绍

官网-https://www.reactive-streams.org/?spm=a2c6h.12873639.0.0.edf277a6wQI9QB

简介:

Reactive Streams是一个对于 异步流处理且伴随非阻塞背压机制 而提供的倡议规范;

目标:

控制异步边界的流数据交换(例如从一个线程池向另一个线程池传递数据),同时要确保接收端不被强迫 缓冲任意数量的数据,也就是利用背压(backpressure)模型调节线程间的队列;

反应式编程的范式(接口规范),主要接口

Java异步实现的N种方式

  • Publisher

  • Subscriber

  • Subcription

其中,Subcriber 中便包含了上面表格提到的 onNext、onError、onCompleted 这三个方法。

Java异步实现的N种方式

一个简单样例

public static void main(String[] args) throws InterruptedException {
    Flux.just(1, 2, 3, 4, 5)
            .subscribeOn(Schedulers.parallel())
            .subscribe(new CoreSubscriber<Integer>() {
                @Override
                public void onSubscribe(Subscription s) {
                    System.out.println(printThread("onSubscribe, " + s.getClass().toString()));
                    s.request(5);
                }

                @Override
                public void onNext(Integer integer) {
                    System.out.println(printThread("next: " + integer));
                }

                @Override
                public void onError(Throwable t) {

                }

                @Override
                public void onComplete() {
                    System.out.println(printThread("complete"));
                }
            });
    Thread.sleep(1000);
}

private static String printThread(String note) {
    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("hh:mm:ss");
    long time = System.currentTimeMillis();
    Date date = new Date(time);
    return Thread.currentThread().getName() + " " + simpleDateFormat.format(date) + " " + time + " " + note;
}

Java异步实现的N种方式

Reactor

实现reactive streams的类库-Reactor 3 Reference Guide

相似的类库有RxJava2, JDK9 Flow等

Mono 实现了 org.reactivestreams.Publisher 接口,代表0到1个元素的发布者。

Flux 同样实现了 org.reactivestreams.Publisher 接口,代表0到N个元素的发表者。

Scheduler 表示背后驱动反应式流的调度器,通常由各种线程池实现。

@Test
    public void ReactorTest() throws InterruptedException {
        System.out.println(printThread("小明点餐"));
        Mono mono = Mono.fromSupplier(() -> {
            System.out.println(printThread("厨师开始做菜"));
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(printThread("厨师菜做好了"));
            return "菜已装盘";
        }).publishOn(Schedulers.parallel())
          .zipWith(Mono.fromSupplier(() -> {
            System.out.println(printThread( "小明开始玩游戏"));
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(printThread("小明结束玩游戏"));
            return " ";
        })).doOnSuccess((tuple2) -> System.out.println(printThread( tuple2.getT1() + ", 小明开始吃饭")));

        mono.subscribe();

        Thread.sleep(10000);
    }

Java异步实现的N种方式

reactor操作函数

Reactor 3 Reference Guide

版权声明:程序员胖胖胖虎阿 发表于 2022年11月2日 上午9:24。
转载请注明:Java异步实现的N种方式 | 胖虎的工具箱-编程导航

相关文章

暂无评论

暂无评论...