Java多线程工具CompletableFuture详解

// 你的下一步业务逻辑
Runnable next = new MyRunnable();
// 因为是 run 所以无返回值， 因为是 then 所以在 completableFuture1 对应的任务结束后，执行一段任务
completableFuture1.thenRun(next);
// 因为是 run 所以无返回值， 因为是 then 所以在 completableFuture1 对应的任务结束后，因为 async ，所以要另起一个线程，执行一段任务
completableFuture1.thenRunAsync(next);

// 自定义线程池
ThreadPoolExecutor executorService = new ThreadPoolExecutor(10, 20,
        2L, TimeUnit.SECONDS,
        new LinkedBlockingDeque<Runnable>(10),
        // 你可以根据你的需要自己实现 Handler，此处为简写使用现成的 Handler，此处捕获的是流量过载异常
        new ThreadPoolExecutor.AbortPolicy());
// 通过自定义线程池使用 CompletableFuture
CompletableFuture.runAsync(() -> {
    System.out.println("我的业务代码");
}, executorService);

public static void sleep(Long sleepTime) {
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

completableFuture.exceptionally((Throwable ex) -> {
            ex.printStackTrace();
			// 取决于 completableFuture 有无返回值，类型是啥。此处实例是 completableFuture 为 CompletableFuture<Long>
            return -1L;
        });

CompletableFuture completableFuture = new CompletableFuture();
    Long startTs = System.currentTimeMillis();
    new Thread(() -> {
		sleep(200L);
		completableFuture.complete("完成");// 设置 completableFuture 结果并将状态设置为已完成
    }).start();
    while (!completableFuture.isDone()) {
        // 非阻塞式获取结果，如果当前未执行完成则返回入参字符串"我还没完成"
        // 执行到此处恰好任务完成，然后在执行 while 循环判断跳出，你在循环内最多输出一次 “完成”
        System.out.println(completableFuture.getNow("未完"));
    }
    System.out.println("最终结果：" + completableFuture.getNow("未完") + " " + (System.currentTimeMillis() - startTs) + " ms");

public static void 阻塞到任意一个完成() throws IOException {
        // 模拟一个耗时 20 L 的任务
        CompletableFuture<Long> completableFuture1 = CompletableFuture.supplyAsync(() -> {
            sleep(20L);
            System.out.println("completableFuture1 完成" + " --" + (System.currentTimeMillis() - startTs));
            return 20L;
        });
        // 模拟一个耗时 10 L 的任务
        CompletableFuture<Long> completableFuture2 = CompletableFuture.supplyAsync(() -> {
            sleep(10L);
            System.out.println("completableFuture2 完成" + " --" + (System.currentTimeMillis() - startTs));
            return 10L;
        });
        // applyToEitherAsync 代表另起一个线程去执行第二个入参的代码块，这里其实没啥影响，我就不加 Async 了
        CompletableFuture result = completableFuture1.applyToEither(completableFuture2, fasterOne -> {
            System.out.println(fasterOne);
            return fasterOne;
        });
        // 除了写回调函数方法外的另一种获取最快值的方法
        System.out.println("最快的为：" + result.join());
        // 调用读取行阻塞住，防止异步任务还未完成就退出了
        System.in.read();
    }

CompletableFuture[] array = new CompletableFuture[2];
        array[0] = completableFuture1;
        array[1] = completableFuture2;
        CompletableFuture fasterOne = CompletableFuture.anyOf(array);
        System.out.println("最快的为：" + fasterOne.join());

public static void 阻塞到全部完成() {
        Long startTs = System.currentTimeMillis();
        // 模拟一个耗时 20 L 的任务
        CompletableFuture<Long> completableFuture1 = CompletableFuture.supplyAsync(() -> {
            sleep(20L);
            System.out.println("completableFuture1 完成" + " --" + (System.currentTimeMillis() - startTs));
            return 20L;
        });
        // 模拟一个耗时 10 L 的任务
        CompletableFuture<Long> completableFuture2 = CompletableFuture.supplyAsync(() -> {
            sleep(10L);
            System.out.println("completableFuture2 完成" + " --" + (System.currentTimeMillis() - startTs));
            return 10L;
        });
        // thenAcceptBothAsync 代表另起一个线程去执行第二个入参的代码块，这里其实没啥影响，我就不加 Async 了
        CompletableFuture result = completableFuture1.thenAcceptBoth(completableFuture2, (r1, r2) -> {
            System.out.println("completableFuture1 ：" + r1);
            System.out.println("completableFuture2 ：" + r2);
        });
        // thenAcceptBoth 是没有返回值的，所以这里是 null ，但这句代码还是有作用的，相当于阻塞到全部任务都完成
        System.out.println("返回这个 null 之后意味着全部任务已完成：" + result.join());
    }

CompletableFuture[] array = new CompletableFuture[2];
    array[0] = completableFuture1;
    array[1] = completableFuture2;
    CompletableFuture<Void> all = CompletableFuture.allOf(array);
    System.out.println("返回这个 null 之后意味着全部任务已完成：" + all.join());

public static void 合并() {
        Long startTs = System.currentTimeMillis();
        // 模拟一个耗时 100 L 的任务
        CompletableFuture<Long> completableFuture1 = CompletableFuture.supplyAsync(() -> {
            sleep(100L);
            System.out.println("completableFuture1 完成" + " --" + (System.currentTimeMillis() - startTs));
            return 100L;
        });
        // 模拟一个耗时 100 L 的任务
        CompletableFuture<Long> completableFuture2 = CompletableFuture.supplyAsync(() -> {
            sleep(100L);
            System.out.println("completableFuture2 完成" + " --" + (System.currentTimeMillis() - startTs));
            return 100L;
        });
        // 模拟 completableFuture1 合并一个耗时 120 L 的任务，返回值为两个任务总工时
        // thenCombineAsync 代表另起一个线程去执行第二个入参的代码块，这里其实没啥影响，我就不加 Async 了
        CompletableFuture<Long> completableFuture3 = completableFuture1.thenCombine(CompletableFuture.supplyAsync(() -> {
            sleep(120L);
            return 120L;
        }), (x, y) -> x + y);
        // 模拟 completableFuture2 合并一个耗时 50 L 的任务，返回值为两个任务总工时
        // thenComposeAsync 代表另起一个线程去执行第一个入参的代码块，这里其实没啥影响，我就不加 Async 了
        CompletableFuture<Long> completableFuture4 = completableFuture2.thenCompose(r2 -> {
            CompletableFuture<Long> temp = CompletableFuture.completedFuture(r2);
            return temp.thenApply(rTemp -> {
                sleep(50L);
                return temp.join() + 50L;
            });
        });
        boolean printFlag3 = true;
        boolean printFlag4 = true;
        String completableFuture3Info = null;
        String completableFuture4Info = null;
        while (!completableFuture3.isDone() || !completableFuture4.isDone()) {
            if (completableFuture3.isDone()) {
                if (printFlag3) {
                    printFlag3 = false;
                    completableFuture3Info = "completableFuture3 完成:" + completableFuture3.join() + " --" + (System.currentTimeMillis() - startTs);
                }
            }
            if (completableFuture4.isDone()) {
                if (printFlag4) {
                    printFlag4 = false;
                    completableFuture4Info = "completableFuture4 完成:" + completableFuture4.join() + " --" + (System.currentTimeMillis() - startTs);
                }
            }
        }
        System.out.println(completableFuture3Info != null ? completableFuture3Info : "completableFuture3 actual：" + completableFuture3.getNow(-100L) + " --" + (System.currentTimeMillis() - startTs));
        System.out.println(completableFuture4Info != null ? completableFuture4Info : "completableFuture4 actual：" + completableFuture4.getNow(-100L) + " --" + (System.currentTimeMillis() - startTs));
    }