【多线程开发 3】从源码到实践CompletableFuture

news2024/11/16 21:30:30

【多线程开发 3】从源码到实践CompletableFuture

2024年5月22日

本文将从以下几个点讲解CompletableFuture：

前身
是什么？
可以用来做什么？
源码原理
实战
其他包装类

前身

CompletableFuture的依赖如图所示：
在这里插入图片描述

在这里插入图片描述

所以我打算从Runnable、Future、RunnableFuture、FutureTask讲起

Runnable

Runnable就是一个带着执行方法的接口，通过实现Runnable可以实现开始一个线程执行对应的无返回值的操作。

Future

Future保存异步计算的结果,可以在我们执行任务时去做其他工作，体现了解耦的思想。Future 可以对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。

其实本质上就是对Runnable或者Callable任务做了一层异步方法封装

RunnableFuture

RunnableFuture实际上只是Runnable接口和Future接口的总和，使用较少。

FutureTask

FutureTask是Future接口的一个唯一实现类。

FutureTask可以对Task做一个异步执行，通过FutureTask的构造函数可以将你想进行异步操作的Task封装进去。FutureTask是个适配器类，作为一个转换器，将Runnable和Callable适配，适配器类是适配器模式的核心，它通过实现RunnableFuture和组合Callable使得两者产生关系。

CompletableFuture可以做什么？

如果只使用Future类可以进行简单的异步执行，但是使用Future的get方法会导致主进程阻塞，并且也无法做到多个异步任务之间的依赖关系，并且有时候异步任务报错不会输出堆栈，Future任务也不会触发后续操作，没有提供函数式编程，不过以上CompletableFuture都可以帮助我们实现。

实战

以下是一些常用API，需要注意的是CompletableFuture可以使用默认线程池或者是自定义线程池，默认线程池是ForkJoinPool

    @Test
    public void test1() throws ExecutionException, InterruptedException, java.util.concurrent.ExecutionException {
        CompletableFuture<String> base = new CompletableFuture<>();
        CompletableFuture<String> future = base.thenApply(s -> s + " 2").thenApply(s -> s + " 3");
        base.complete("1");
        System.out.println(future.get());
    }


    @Test
    public void test2() throws ExecutionException, InterruptedException, java.util.concurrent.ExecutionException {
        CompletableFuture<String> base = new CompletableFuture<>();
        CompletableFuture<String> future = base.thenApply(s -> s + " 2").thenApply(s -> s + " 3");
        future.complete("1");
        System.out.println(future.get());
    }


    @Test
    public void test3() throws java.util.concurrent.ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        CompletableFuture<String> step1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行【步骤1】");
            return "【步骤1的执行结果】";
        }, executor);
        CompletableFuture<String> step2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行【步骤2】");
            return "【步骤2的执行结果】";
        }, executor);
        CompletableFuture<String> step3 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行【步骤3】");
            return "【步骤3的执行结果】";
        }, executor);

        CompletableFuture<Object> stepN = CompletableFuture.anyOf(step1, step2, step3);
        System.out.println("步骤N的结果：" + stepN.get());

        executor.shutdown();
    }

    @Test
    public void test4() throws java.util.concurrent.ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        CompletableFuture<String> step1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行【步骤1】");
            return "【步骤1的执行结果】";
        }, executor);
        CompletableFuture<String> step2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行【步骤2】");
            return "【步骤2的执行结果】";
        }, executor);
        CompletableFuture<String> step3 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行【步骤3】");
            return "【步骤3的执行结果】";
        }, executor);

        CompletableFuture<Void> stepN = CompletableFuture.allOf(step1, step2, step3);
        System.out.println("步骤N的结果：" + stepN.get());

        executor.shutdown();
    }

    @Test
    public void test5() throws java.util.concurrent.ExecutionException, InterruptedException {

        // 生产者，可以指定返回结果
        CompletableFuture<String> firstTask = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务开始执行");
            System.out.println("异步任务执行结束");
            return "返回结果";
        });

        String result1 = firstTask.join();
        String result2 = null;
        try {
            result2 = firstTask.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(result1 + "," + result2);
    }


    @Test
    public void test6() throws IOException {
        CompletableFuture.supplyAsync(() -> {
            System.out.println("任务A");
            return "abcdefg";
        }).thenAccept(result -> {
            System.out.println("任务b，拿到结果处理：" + result);
        });

        System.in.read();

    }

    @Test
    public void test7() throws IOException {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务A执行");
            return 10;
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            System.out.println("任务B执行");
            return 10;
        }), (r1, r2) -> {
            System.out.println("任务C执行");
            return r1 + r2;
        });
        System.out.println("任务C结果=" + future.join());
    }