发布于 由admin

运用 Java CompletableFuture：编写异步非阻塞代码，提高程序的并发处理能力。

好嘞，各位看官老爷们，今天咱就来聊聊Java世界里的一位重量级选手——CompletableFuture。这家伙可不是吃素的，它能让我们的代码跑得飞起，并发能力蹭蹭往上涨，简直是提高程序性能的秘密武器！准备好迎接这场异步非阻塞的盛宴了吗？😎

第一幕：同步世界的烦恼

在深入CompletableFuture的妙处之前，咱们先回顾一下同步编程的老路子。想想那些年，我们写的代码就像老老实实的排队，一个任务没完成，下一个就得等着。

public class SyncExample {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("开始执行任务...");
        String result1 = task1(); // 耗时任务1
        System.out.println("任务1结果: " + result1);
        String result2 = task2(result1); // 耗时任务2，依赖任务1的结果
        System.out.println("任务2结果: " + result2);
        System.out.println("任务全部完成！");
    }

    static String task1() throws InterruptedException {
        Thread.sleep(2000); // 模拟耗时操作
        return "任务1完成";
    }

    static String task2(String input) throws InterruptedException {
        Thread.sleep(1000); // 模拟耗时操作
        return "任务2完成，基于" + input;
    }
}

运行上面的代码，你会发现它慢吞吞的，像蜗牛爬山一样。这是因为 task1 完成后，task2 才能开始，中间的等待时间白白浪费了。这种“你等我，我等你”的模式，在并发量大的时候简直是噩梦！想象一下，如果每个用户请求都这么慢，服务器早就崩溃了。🤯

第二幕：异步非阻塞的曙光

这时候，CompletableFuture就如同救世主般闪亮登场了！它带来了异步非阻塞的编程模式，让我们的代码可以“一心多用”，不用傻傻等待，而是可以先去处理其他任务，等结果准备好了再回来处理。

什么是异步非阻塞？

  • 异步 (Asynchronous): 不用立即拿到结果，而是先发起请求，等结果准备好后，系统会通知你。就像你点外卖，不用一直守着，外卖小哥到了会给你打电话。
  • 非阻塞 (Non-blocking): 发起请求后，不用一直等待，可以去做其他事情。就像你打电话，不用一直拿着电话等对方接听，可以先挂断，等对方回过来。

第三幕：CompletableFuture的华丽登场

CompletableFuture是Java 8引入的一个强大的类，它实现了 Future 接口和 CompletionStage 接口，提供了丰富的异步编程方法。

CompletableFuture的核心方法：

方法 描述
supplyAsync(Supplier<U> supplier) 异步执行一个有返回值的任务，使用默认线程池。
supplyAsync(Supplier<U> supplier, Executor executor) 异步执行一个有返回值的任务，使用指定的线程池。
runAsync(Runnable runnable) 异步执行一个没有返回值的任务，使用默认线程池。
runAsync(Runnable runnable, Executor executor) 异步执行一个没有返回值的任务，使用指定的线程池。
thenApply(Function<? super T,? extends U> fn) 当前任务完成后，将结果传递给 fn 函数，执行下一步操作，返回一个新的 CompletableFuture。 （同步执行）
thenApplyAsync(Function<? super T,? extends U> fn) 当前任务完成后，将结果传递给 fn 函数，异步执行下一步操作，返回一个新的 CompletableFuture。 （异步执行）
thenAccept(Consumer<? super T> action) 当前任务完成后，将结果传递给 action 消费，没有返回值。 （同步执行）
thenAcceptAsync(Consumer<? super T> action) 当前任务完成后，将结果传递给 action 消费，异步执行，没有返回值。 （异步执行）
thenRun(Runnable action) 当前任务完成后，执行 action，不关心当前任务的结果，没有返回值。 （同步执行）
thenRunAsync(Runnable action) 当前任务完成后，异步执行 action，不关心当前任务的结果，没有返回值。 （异步执行）
thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) 将两个 CompletionStage 的结果合并，执行 fn 函数，返回一个新的 CompletableFuture。 （同步执行）
thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) 将两个 CompletionStage 的结果合并，异步执行 fn 函数，返回一个新的 CompletableFuture。 （异步执行）
thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) 当前任务完成后，将结果传递给 fn 函数，fn 函数返回一个新的 CompletionStage，相当于将两个 CompletionStage 串联起来。 （同步执行）
thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) 当前任务完成后，将结果传递给 fn 函数，fn 函数返回一个新的 CompletionStage，相当于将两个 CompletionStage 串联起来。 （异步执行）
exceptionally(Function<Throwable, ? extends T> fn) 当任务发生异常时，执行 fn 函数，返回一个备用结果。
handle(BiFunction<? super T, Throwable, ? extends U> fn) 无论任务正常完成还是发生异常，都执行 fn 函数，可以用来处理异常或返回默认值。
whenComplete(BiConsumer<? super T, ? super Throwable> action) 无论任务正常完成还是发生异常，都执行 action，可以用来记录日志或释放资源。
join() 等待任务完成并返回结果，如果任务发生异常，会抛出 CompletionException。
get() 等待任务完成并返回结果，如果任务发生异常，会抛出 ExecutionException。
get(long timeout, TimeUnit unit) 等待任务完成并返回结果，如果在指定时间内没有完成，会抛出 TimeoutException。
cancel(boolean mayInterruptIfRunning) 取消任务的执行。
isDone() 判断任务是否完成。

第四幕：CompletableFuture实战演练

咱们用CompletableFuture来改造一下之前的同步例子：

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class AsyncExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("开始执行任务...");

        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000); // 模拟耗时操作
                return "任务1完成";
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        CompletableFuture<String> future2 = future1.thenApplyAsync(result1 -> {
            try {
                Thread.sleep(1000); // 模拟耗时操作
                return "任务2完成，基于" + result1;
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        // 主线程可以继续做其他事情，不用等待任务完成
        System.out.println("主线程可以做其他事情了...");

        // 获取任务结果
        String result2 = future2.get();
        System.out.println("任务2结果: " + result2);
        System.out.println("任务全部完成！");
    }
}

在这个例子中，我们使用 supplyAsync 异步执行 task1，并使用 thenApplyAsync 在 task1 完成后异步执行 task2。主线程不用等待，可以继续执行其他任务。当需要结果时，再调用 get() 方法获取。

关键点：

  • supplyAsync：异步执行一个Supplier接口的任务，并返回一个CompletableFuture对象。
  • thenApplyAsync：当第一个CompletableFuture完成后，将结果传递给下一个Function接口的任务，并异步执行，返回一个新的CompletableFuture对象。
  • get()：阻塞等待CompletableFuture完成，并返回结果。

第五幕：CompletableFuture的各种姿势

CompletableFuture的功能远不止这些，它还提供了很多其他的操作，让我们来看看一些常用的姿势：

  1. 组合多个CompletableFuture：

    有时候，我们需要等待多个任务都完成后才能进行下一步操作。CompletableFuture提供了 allOf 和 anyOf 方法来实现这个功能。

    • allOf(CompletableFuture<?>... cfs)：等待所有任务完成。
    • anyOf(CompletableFuture<?>... cfs)：只要有一个任务完成。
    CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1500);
        return "任务3完成";
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
});

CompletableFuture.allOf(future1, future3).thenRun(() -> {
    System.out.println("任务1和任务3都完成了！");
});

  2. 处理异常：

    异步任务可能会出现异常，CompletableFuture提供了 exceptionally 和 handle 方法来处理异常。

    • exceptionally(Function<Throwable, ? extends T> fn)：当任务发生异常时，执行 fn 函数，返回一个备用结果。
    • handle(BiFunction<? super T, Throwable, ? extends U> fn)：无论任务正常完成还是发生异常，都执行 fn 函数，可以用来处理异常或返回默认值。
    CompletableFuture<String> future4 = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("任务4发生异常！");
}).exceptionally(ex -> {
    System.out.println("任务4发生异常: " + ex.getMessage());
    return "任务4完成 (备用结果)";
});

  3. 使用线程池：

    CompletableFuture默认使用 ForkJoinPool.commonPool() 作为线程池，但我们可以使用 Executor 来指定自定义的线程池。

    import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

ExecutorService executor = Executors.newFixedThreadPool(10);

CompletableFuture<String> future5 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
        return "任务5完成";
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}, executor);

第六幕：CompletableFuture的优势与适用场景

CompletableFuture的优势：

  • 提高并发能力： 异步非阻塞的模式可以让程序同时处理多个任务，提高并发能力。
  • 提高响应速度： 不用等待耗时任务完成，可以先响应用户请求，提高用户体验。
  • 简化异步编程： 提供了丰富的API，简化了异步编程的复杂度。

CompletableFuture的适用场景：

  • IO密集型任务： 例如网络请求、数据库查询等，可以利用异步非阻塞的特性，避免线程阻塞。
  • 需要并发处理多个任务的场景： 例如批量处理数据、并行执行多个算法等。
  • 需要组合多个异步任务的场景： 例如一个请求需要调用多个微服务，并将结果合并。

第七幕：CompletableFuture的注意事项

在使用CompletableFuture时，也要注意以下几点：

  • 避免过度使用： 异步编程会增加代码的复杂度，不要为了异步而异步，只有在确实需要提高并发能力时才使用。
  • 合理选择线程池： 根据任务的类型选择合适的线程池，避免线程池阻塞或资源浪费。
  • 处理异常： 异步任务可能会出现异常，要及时处理，避免程序崩溃。
  • 避免阻塞： 尽量避免在CompletableFuture中使用阻塞操作，例如 Thread.sleep() 或同步IO操作，否则会影响并发能力。
  • 小心死锁： 在使用多个CompletableFuture时，要注意避免死锁，例如互相等待对方完成。

第八幕：CompletableFuture的进阶之路

如果你想更深入地了解CompletableFuture，可以学习以下内容：

  • CompletionStage接口： CompletableFuture实现了CompletionStage接口，可以组合多个CompletionStage，形成复杂的异步流程。
  • Reactive Streams： CompletableFuture可以和Reactive Streams结合使用，构建响应式系统。
  • Project Reactor： Project Reactor是一个基于Reactive Streams的响应式编程框架，可以和CompletableFuture一起使用。

第九幕：总结

CompletableFuture是Java中一个非常强大的异步编程工具，它可以帮助我们编写高效、可伸缩的并发程序。掌握CompletableFuture的使用，可以让你在面试中脱颖而出，也可以让你在工作中游刃有余。记住，CompletableFuture不是银弹，要根据实际情况选择合适的并发模型。希望这篇文章能帮助你打开异步编程的大门，让你的代码跑得更快，飞得更高！🚀

最后，送给大家一句鸡汤：编程就像人生，充满了异步和阻塞，我们要学会优雅地处理它们，才能走向成功！💪

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注