Java 异步编程

发表于 2024-01-30 更新于 2025-01-06 分类于 Java 阅读次数： Waline：本文字数： 2.3k 阅读时长 ≈ 8 分钟

通常，Java开发者使用阻塞式（blocking）同步方式编写代码。这没有问题，在出现性能瓶颈后，我们可以增加处理线程，线程中同样是阻塞的代码。但是这种使用资源的方式会迅速面临资源竞争和并发问题。

更糟糕的是，阻塞会浪费资源。具体来说，比如当一个程序面临延迟（通常是I/O方面，比如数据库读写请求或网络调用），所在线程需要进入 idle 状态等待数据，从而浪费资源。

异步编程适用于大多数IO密集型的系统, 可以提高系统整体的响应性和吞吐量。

Future

在Jdk1.5后, 我们可以用Future来访问异步计算的结果：

Future<Integer> future = executor.submit(() -> {
   // 进行一些耗时计算
   return result;
});

// ...做些其他事情
int result = future.get();  // 获取结果

但是这种方式也有缺陷:

无法方便得知任务何时完成
获得任务结果会导致线程阻塞
缺少异常处理

CompletableFuture 是 Jdk 8 引入的一个类，用于表示异步计算的结果。它提供了一些方法，可以在计算完成后执行一些操作，或者组合多个 CompletableFuture 来实现复杂的异步逻辑。

考虑这样一种情景：在用户界面上显示用户的5个收藏，如果没有任何收藏则提供5个建议。这需要3个服务（一个提供收藏的ID列表，第二个服务获取收藏内容，第三个提供建议内容）：

// 从 userService 获取收藏的 ID 列表
CompletableFuture<List<String>> favorites = userService.getFavorites(userId);

// 从 favoriteService 获取每个收藏的详情
CompletableFuture<List<Detail>> details = favorites.thenCompose(favList -> {
    List<CompletableFuture<Detail>> detailFutures = favList.stream()
            .map(favoriteService::getDetails)
            .collect(Collectors.toList());
    return CompletableFuture.allOf(detailFutures.toArray(new CompletableFuture[0]))
            // 将合并的 future 转换为一个详情列表
            .thenApply(v -> detailFutures.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList()));
});

// 使用 exceptionally 处理 details future 失败或返回空列表的情况
// 在这种情况下，获取一个 CompletableFuture，包含建议内容
CompletableFuture<List<Detail>> suggestions = details.exceptionally(ex -> {
    return suggestionService.getSuggestions();
});

// 将列表项目数量限制为 5 个
CompletableFuture<List<Detail>> topFive = suggestions.thenApply(list -> {
    return list.subList(0, Math.min(5, list.size()));
});

// 在 UI 线程调度器上发布结果，并在 uiList 上显示
topFive.thenAcceptAsync(uiList::show, UiUtils.uiThreadScheduler())
       .exceptionally(ex -> {
           UiUtils.errorPopup(ex);
           return null;
       });

回调函数

提交了一次请求, 然后在事情完成后, 自动"调用回来"执行下一步操作

doSomeAsyncTask(result -> {
   // 耗时任务完成后,执行此回调
   showResult(result); 
});

但是在一些复杂场景会导致“回调地狱”, 导致代码难以理解和维护。比如用Callback实现上面的例子：

userService.getFavorites(userId, new Callback<List<String>>() {
    public void onSuccess(List<String> list) {
        if (list.isEmpty()) {
            // 收藏为空, 获取建议内容
            suggestionService.getSuggestions(new Callback<List<Favorite>>() {
                public void onSuccess(List<Favorite> list) {
                    UiUtils.submitOnUiThread(() -> {
                        list.stream()
                                .limit(5)
                                .forEach(uiList::show);
                    });
                }

                public void onError(Throwable error) {
                    UiUtils.errorPopup(error);
                }
            });
        } else {
            // 获取收藏内容
            list.stream()
                    .limit(5)
                    .forEach(favId -> favoriteService.getDetails(favId,
                            new Callback<Favorite>() {
                                public void onSuccess(Favorite details) {
                                    UiUtils.submitOnUiThread(() -> uiList.show(details));
                                }

                                public void onError(Throwable error) {
                                    UiUtils.errorPopup(error);
                                }
                            }
                    ));
        }
    }

    public void onError(Throwable error) {
        UiUtils.errorPopup(error);
    }
});

响应式

响应式编程说明

响应式编程是强调异步的、非阻塞的、事件驱动的、基于回调或流的编程风格。响应式编程的核心思想是将数据和事件作为流来处理，而不是作为单个的值或对象。开发者可以使用各种操作符来对流进行转换、过滤、合并、分组等操作，从而实现复杂的业务逻辑。

目前，在 Java 领域实现了响应式编程的技术有 Spring 的 Project Reactor、Netflix RxJava 1/2 等。前者的 3.0 版本作为 Spring 5 的基础，在17年底发布，推动了后端领域响应式编程的发展。后者出现时间更早，在前端开发领域应用的比后端更要广泛一些。

上面的例子使用Reactor实现:

userService.getFavorites(userId)  // 获取收藏的ID列表到集合数据流
           .flatMap(favoriteService::getDetails) // 获取收藏内容
           .switchIfEmpty(suggestionService.getSuggestions()) // 如果为空获取建议内容
           .take(5) // 只拿执行流中的5个
           .publishOn(UiUtils.uiThreadScheduler()) // 发布到 UI 线程中进行处理
           .subscribe(uiList::show, UiUtils::errorPopup);  // 订阅处理结果

可以发现响应式版本的代码结构更清晰, 易读性也更高。此外 Reactor 支持背压机制，可以对上游发送反馈信号，控制数据流的速度，避免生产者和消费者之间的压力差异。

缺点是使用了函数式编程和链式调用，需要学习新的 API 和编程模型，以及需要理解和调试多个异步事件和数据流之间的动态交互，学习曲线较陡峭，调试也比较困难。另外虽然相对于 Callback，代码可读性更高，但这种优点是相对的，相对于传统代码，可读性就成了反应式编程的缺点。

R2DBC pk JDBC 和 WebFlux pk Web MVC 评测数据

Project Loom

Pasted image 20240130154036.png

为了让简单和高并发这两个目标兼得，我们需要 Project Loom。

这篇文档介绍了发起 Project Loom 的原因，以及 Java 线程基础的很多底层设计

使用虚拟线程实现上面的例子:

Thread.startVirtualThread(() -> {
    List<String> favorites = userService.getFavorites(userId); // 获取收藏列表

    if (favorites.isEmpty()) { // 收藏为空，获取建议内容
        // 获取建议列表
        List<Favorite> suggestions = suggestionService.getSuggestions()
                .stream()
                .limit(5)
                .toList();
        UiUtils.submitOnUiThread(() -> {
            suggestions.forEach(uiList::show); // 在 UI 线程中显示建议列表
        });
    } else { // 获取收藏内容
        favorites.stream()
            .limit(5)
            .forEach(favId -> {
                Favorite details = favoriteService.getDetails(favId); // 获取收藏详情
                UiUtils.submitOnUiThread(() -> uiList.show(details)); // 在 UI 线程中显示收藏详情
            });
    }
});

Project Loom 是一个 Java 项目，不同于之前的方案，Project Loom 是从 JVM 层面对多线程技术进行彻底的改变。它为开发者提供了一种轻量级的并发抽象，称为虚拟线程，本质上它是一种有栈协程。

Pasted image 20240130154047.png

Green Thread是在OS Thread在用户空间的模拟，是一种在根本没有本机多线程支持的系统上支持多线程的解决方案。在这样的系统中，单个阻塞调用将阻塞整个进程，从而阻塞所有线程。现在已经不常使用了。

虚拟线程有以下几个优势：

虚拟线程不依赖于 OS 线程，而是由 JVM 在内部调度和执行。它只包含 Java 的调用栈，可以在需要的时候暂停和恢复自己的执行状态。
虚拟线程的调用栈可以动态地增长和缩小，节省内存空间和提高性能。相比之下，OS 线程的调用栈是固定的大小，占用大量的内存空间（64 位 Linux 上 HotSpot 的线程栈容量默认是 1MB，而虚拟线程仅需要200-300字节来存储元数据）。因此，虚拟线程可以支持更高的并发度和更低的上下文切换开销。
虚拟线程的数量只受限于可用的内存和 CPU 资源，而不是 OS 的限制。在Project Loom C5M中，已经成功地使用虚拟线程在C/S应用中实现500万个长连接。这意味着虚拟线程可以让开发者使用一对一的模型来编写并发代码，即每个请求或任务对应一个虚拟线程，而不需要使用线程池或其他复杂的机制来复用或管理线程。
虚拟线程可以实现对阻塞操作的透明化。当虚拟线程遇到一个阻塞操作，比如等待 I/O，JVM 会检测到这个操作，并把虚拟线程设置为 park 状态，等待 I/O 响应。同时，JVM 会把 OS 线程分配给其他可运行的虚拟线程，从而避免浪费 CPU 资源。当 I/O 响应到来时，JVM 会把虚拟线程恢复为 run 状态，并重新分配 OS 线程给它。

基于上面的优势，开发者可以使用传统的同步和命令式的编程风格来写异步和非阻塞的代码，而不需要使用复杂的回调或者函数式编程。代码变得更简单、更清晰、更容易调试和维护，并发性能和资源利用率也得以提高。

很快，Project Loom 将在 JDK 21 中正式发布。JDK 21 也是下一个长期支持（LTS）的版本，这意味着 Project Loom 将得到更广泛和更稳定的应用。Project Loom 的出现可以使得 Java 在云原生时代能更好地适应高并发、低延迟、高吞吐的需求，同时简化开发者的编程模型和心智负担，让 Java 摆脱传统的线程模型的束缚，提高资源利用率和性能，降低复杂度和成本。Project Loom 无疑是 Java 的一个重大创新，它将为 Java 的未来带来更多的可能性和机遇。

参考

Java 异步编程：从 Future 到 Loom-今日头条 (toutiao.com)
云原生时代的Java (geekbang.org)
java - Project loom: what makes the performance better when using virtual threads? - Stack Overflow
如何看待Project Loom? - 圆胖肿的回答 - 知乎
对于后端开发，响应式编程真的是大势所趋吗？ - 知乎 (zhihu.com)