jdk8特性-CompletableFuture

  • CompletionStage接口翻译成中文是“完工阶段“,是java8新增的一个工具。定义的一系列方法,接收的参数有有三类CompletionStage,Runnable,Consumer和Function方法。面向函数式编程,因为入参抽象的是方法,并且将结果传入下一个被调用的方法。该接口表示异步计算的某种状态。到了这个状态会出发对应的方法。

  • CompletableFuture分别实现了Fulture和CompletionStage接口。也就是说拥有异步计算的能力,通过回调来处理计算的结果。当然包括了多个CompletableFuture之间的组合。增加了多线程之间的协调性。可以是完成了也可以是未完成的某个阶段。这样的组合提供了很多编程的可能性。

  • 话不多说,我们直接上代码,用法都总结和归类到写在注释上

package com.example.demo;

import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

/**
 * Project <demo-project>
 * Created by jorgezhong on 2018/9/8 11:45.
 */
public class CompletableFutureDemo {


    /**
     * 创建CompletableFuture
     * - runAsync
     * - supplyAsync
     * - completedFuture
     * <p>
     * 异步计算启用的线程池是守护线程
     */
    @Test
    public void test1() {

        //1、异步计算:无返回值

        //默认线程池为:ForkJoinPool.commonPool()
        CompletableFuture.runAsync(() -> {
            // TODO: 2018/9/8 无返回异步计算
            System.out.println(Thread.currentThread().isDaemon());
        });

        //指定线程池,(到了jdk9CompletableFuture还拓展了延迟的线程池)
        CompletableFuture.runAsync(() -> {
            // TODO: 2018/9/8 无返回异步计算
        }, Executors.newFixedThreadPool(2));


        //2、异步计算:有返回值

        // 使用默认线程池
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "result1");
        //getNow指定异步计算抛出异常或结果返回null时替代的的值
        String result1 = future1.getNow(null);


        //  指定线程池
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "result2", Executors.newFixedThreadPool(2));
        //getNow指定异步计算抛出异常或结果返回null时替代的的值
        String result2 = future2.getNow(null);


        //3、初始化一个有结果无计算的CompletableFuture
        CompletableFuture<String> future = CompletableFuture.completedFuture("result");
        String now = future.getNow(null);
        System.out.println("now = " + now);


    }


    /**
     * 计算完成时需要对异常进行处理或者对结果进行处理
     * - whenComplete:同步处理包括异常
     * - thenApply:同步处理正常结果(前提是没有异常)
     * <p>
     * - whenCompleteAsync:异步处理包括异常
     * - thenApplyAsync:异步处理正常结果(前提是没有异常)
     * <p>
     * - exceptionally : 处理异常
     */
    @Test
    public void test2() {


        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "result");

        //whenComplete方法收future的结果和异常,可灵活进行处理
        //1、同步处理

        //  无返回值:可处理异常
        future.whenComplete((result, throwable) -> System.out.println("result = " + result));

        //  有返回值:没有异常处理(前提)
        CompletableFuture<String> resultFuture1 = future.thenApply(result -> "result");
        String result1 = resultFuture1.getNow(null);


        //2、异步处理:

        //  无返回值: 默认线程池
        future.whenCompleteAsync((result, throwable) -> System.out.println("result = " + result));
        //  无返回值:指定线程池
        future.whenCompleteAsync((result, throwable) -> System.out.println("result = " + result), Executors.newFixedThreadPool(2));

        //  有返回值:默认线程池
        CompletableFuture<String> resultFuture2 = future.thenApplyAsync(result -> "result");
        String result2 = resultFuture2.getNow(null);

        //  有返回值:指定线程池
        CompletableFuture<String> resultFuture3 = future.thenApplyAsync(result -> "result", Executors.newFixedThreadPool(2));
        String result3 = resultFuture3.getNow(null);


        //3、处理异常,处理完之后返回一个结果
        CompletableFuture<String> exceptionallyFuture = future.whenCompleteAsync((result, throwable) -> System.out.println("result = " + 1 / 0))
                .exceptionally(throwable -> "发生异常了:" + throwable.getMessage());
        System.out.println(exceptionallyFuture.getNow(null));


    }


    /**
     * 异常处理还可以使用以下两个方法
     * - handle
     * - handleAsync
     * <p>
     * 备注:exceptionally同步和异步计算一起用如果出现异常会把异常抛出。用以上的方法可以拦截处理
     */
    @Test
    public void test3() {


        CompletableFuture<String> exceptionoHandle = CompletableFuture.completedFuture("produce msg")
                .thenApplyAsync(s -> "result" + 1 / 0);

        String handleResult1 = exceptionoHandle.handle((s, throwable) -> {
            if (throwable != null) {
                return throwable.getMessage();
            }
            return s;
        }).getNow(null);

        //指定线程池
        String handleResult2 = exceptionoHandle.handleAsync((s, throwable) -> {
            if (throwable != null) {
                return throwable.getMessage();
            }
            return s;
        }, Executors.newFixedThreadPool(2)).getNow(null);

    }

    /**
     * 生产--消费
     * - thenAccept:同步的
     * - thenAcceptAsync:异步的
     * <p>
     * 接受上一个处理结果,并实现一个Consumer,消费结果
     */
    @Test
    public void test4() {

        //同步的
        CompletableFuture.completedFuture("produce msg")
                .thenAccept(s -> System.out.println("sync consumed msg : " + s));

        //异步的
        //默认线程池
        CompletableFuture.completedFuture("produce msg")
                .thenAcceptAsync(s -> System.out.println("async consumed msg : " + s));
        //指定线程池
        CompletableFuture.completedFuture("produce msg")
                .thenAcceptAsync(s -> System.out.println("async consumed msg : " + s), Executors.newFixedThreadPool(2));
    }


    /**
     * 取消任务
     * - cancel
     */
    @Test
    public void test5() throws InterruptedException {

        CompletableFuture<String> message = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {

            try {
                Thread.sleep(800);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return s + "result";

        });
        String now = message.getNow(null);
        System.out.println("now = " + now);

        //取消
        boolean cancel = message.cancel(true);
        System.out.println("cancel = " + cancel);


        //如果这里再去获取,会抛出异常,说明已经取消了
        //String now1 = message.getNow(null);

        Thread.sleep(1000);

    }


    /**
     * 两个异步计算
     * - applyToEither:有返回值,同步
     * - acceptEither:无返回值,同步
     * - applyToEitherAsync:有返回值,异步
     * -
     */
    @Test
    public void test6() {


        CompletableFuture<String> task1 = CompletableFuture.completedFuture("task1")
                .thenApply(s -> "task1的计算结果:s1 = " + s);

        //同步,有返回值
        //applyToEither第二个参数接收的值是task1计算的返回值
        CompletableFuture<String> result1 = task1.applyToEither(CompletableFuture.completedFuture("task2")
                .thenApply(s -> "task2的计算结果:s2 = " + s), s -> s);
        System.out.println("task2:" + result1.getNow(null));


        //同步,无返回值
        task1.acceptEither(CompletableFuture.completedFuture("task3")
                .thenApply(s -> "task3的计算结果:s3 = " + s), s -> System.out.println("task3:" + s));


        //异步有返回值,默认线程池,也可以指定
        CompletableFuture<String> result2 = task1.applyToEitherAsync(CompletableFuture.completedFuture("task4")
                .thenApply(s -> "task4的计算结果:s4 = " + s), s -> s);
        //由于是异步的,主线程跑的快一点,因此join()之后才能看到跑完的结果
        System.out.println("task4:" + result2.join());


        //异步无返回值,指定线程池,也可以使用默认线程池
        CompletableFuture<Void> task5 = task1.acceptEitherAsync(CompletableFuture.completedFuture("task5")
                .thenApply(s -> "task5的计算结果:s5 = " + s), s -> System.out.println("task5:" + s), Executors.newFixedThreadPool(2));
        task5.join();


    }

    /**
     * 组合计算结果
     * - runAfterBoth:都计算完之后执行一段代码
     * - thenAcceptBoth:都计算完之后把结果传入,并执行一段代码
     * <p>
     * - thenCombine:组合两个结果
     * - thenCompose:组合两个结果
     */
    @Test
    public void test7() {

        //runAfterBoth方式
        StringBuilder msg = new StringBuilder("jorgeZhong");
        CompletableFuture.completedFuture(msg)
                .thenApply(s -> s.append(" task1,"))
                .runAfterBoth(CompletableFuture.completedFuture(msg)
                        .thenApply(s -> s.append(" task2")), () -> System.out.println(msg));


        //thenAcceptBoth方式
        CompletableFuture.completedFuture("jorgeZhong")
                .thenApplyAsync(String::toLowerCase)
                .thenAcceptBoth(CompletableFuture.completedFuture("jorgeZhong")
                        .thenApplyAsync(String::toUpperCase), (s, s2) -> System.out
                        .println("s1:" + s + ", s2:" + s2));


        //thenCombine方式
        CompletableFuture<String> result1 = CompletableFuture.completedFuture("jorgeZhong")
                .thenApply(String::toLowerCase)
                .thenCombine(CompletableFuture.completedFuture("jorgeZhong")
                        .thenApply(String::toUpperCase), (s, s2) -> "s1:" + s + ", s2:" + s2);

        System.out.println("result1:" + result1.getNow(null));

        //异步
        CompletableFuture<String> result11 = CompletableFuture.completedFuture("jorgeZhong")
                .thenApply(String::toLowerCase)
                .thenCombineAsync(CompletableFuture.completedFuture("jorgeZhong")
                        .thenApplyAsync(String::toUpperCase), (s, s2) -> "s1:" + s + ", s2:" + s2);
        System.out.println("result11:" + result11.join());


        //thenCompose方式
        CompletableFuture<String> result2 = CompletableFuture.completedFuture("jorgeZhong")
                .thenApply(String::toLowerCase)
                .thenCompose(s -> CompletableFuture.completedFuture("jorgeZhong")
                        .thenApply(String::toUpperCase)
                        .thenApply(s1 -> "s:" + s + ", s1:" + s1));
        System.out.println("result2:" + result2.getNow(null));

        //异步
        CompletableFuture<String> result22 = CompletableFuture.completedFuture("jorgeZhong")
                .thenApply(String::toLowerCase)
                .thenComposeAsync(s -> CompletableFuture.completedFuture("jorgeZhong")
                        .thenApplyAsync(String::toUpperCase)
                        .thenApplyAsync(s1 -> "s:" + s + ", s1:" + s1));

        System.out.println("result22:" + result22.join());
    }


    /**
     * 多个CompletableFuture策略
     * - anyOf:接受一个CompletableFuture数组,任意一个任务执行完返回。都会触发该CompletableFuture
     * - whenComplete:计算执行完之后执行实现的一段代码,将上一个结果和异常作为参数传入
     */
    @Test
    public void test8() throws InterruptedException {

        List<String> messages = Arrays.asList("a", "b", "c");
        CompletableFuture.anyOf(messages.stream()
                .map(o -> CompletableFuture.completedFuture(o).thenApplyAsync(s -> {

                    try {
                        Thread.sleep(new Random().ints(99, 300).findFirst().getAsInt());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return s.toUpperCase();
                }))
                .toArray(CompletableFuture[]::new))
                .whenComplete((res, throwable) -> {
                    if (throwable == null) {
                        System.out.println(res.toString());
                    }
                });


        Thread.sleep(1000);


    }


    /**
     * 多个CompletableFuture策略
     * - allOf:接受一个CompletableFuture数组,所有任务返回后,创建一个CompletableFuture
     */
    @Test
    public void test9() {

        List<String> messages = Arrays.asList("a", "b", "c");
        CompletableFuture[] cfs = messages.stream()
                .map(s -> CompletableFuture.completedFuture(s).thenApplyAsync(String::toUpperCase))
                .toArray(CompletableFuture[]::new);
        CompletableFuture.allOf(cfs)
                .whenCompleteAsync((aVoid, throwable) -> Arrays.stream(cfs).forEach(completableFuture -> System.out
                        .println(completableFuture.getNow(null))));


    }


}


上一篇:人工智能真的有那么神秘么,推荐一份机器学习入门书单搞定它!


下一篇:VDI序曲二十三 制作OFFICE 2003应用程序虚拟化序列