将虚拟线程（Project Loom）与 Spring WebFlux/Reactor/Reactive 库结合使用

Question

Java虚拟线程

Java 19 中引入了虚拟线程JEP-425作为预览功能。

在对 Java 虚拟线程（Project Loom）的概念（有时称为轻量级线程（有时称为纤维或绿色线程））进行一些研究之后，我对它们与反应式库的潜在使用非常感兴趣，例如基于 Spring WebFlux在 Project Reactor（反应流实现）和 Netty 上，用于有效地进行阻塞调用。

如今，大多数 JVM 实现都将 Java 线程实现为操作系统线程的瘦直接包装器，有时称为重量级、操作系统管理的线程平台线程。

虽然平台线程一次只能执行一个线程，但是当当前执行的虚拟线程进行阻塞调用（例如网络、文件系统、数据库调用）时，虚拟线程能够切换到执行不同的虚拟线程。

我们如何处理 Reactor 中的阻塞调用？

因此，在 Reactor 中处理阻塞调用时，我们使用以下构造：

Mono.fromCallable(() -> {
     return blockingOperation();
}).subscribeOn(Schedulers.boundedElastic());

我们subcribeOn()提供了一个Scheduler创建专用线程来执行该阻塞操作的方法。然而，这意味着线程最终将被阻塞，因此，由于我们仍然使用老式的线程模型，我们实际上会阻塞平台线程，这仍然不是处理 CPU 资源的真正有效的方式。

这是问题：

所以，问题是，我们是否可以直接使用具有反应式框架的虚拟线程来进行这样的阻塞调用，例如使用Executors.newVirtualThreadPerTaskExecutor()：

创建一个执行器，为每个任务启动一个新的虚拟线程。Executor创建的线程数量是无限制的。

Mono.fromCallable(() -> {
    return blockingOperation();
}).subscribeOn(Schedulers.fromExecutorService(Executors.newVirtualThreadPerTaskExecutor()));

它开箱即用吗？在更有效地处理 CPU 资源并提高应用程序性能方面，我们真的能从这种方法中获益吗？这是否意味着我们可以轻松地将反应式库与任何阻塞库/框架集成，例如 Spring Data JPA（基于 JDBC）和数百万其他库/框架，并神奇地将它们转变为非阻塞？

Answer 1

反应式代码中的阻塞

您也可以阻止反应式代码，但这通常不是一个好主意。

如果没有虚拟线程，执行阻塞操作也会阻塞平台线程，因此如果您阻塞反应式代码，那么您实际上会浪费平台线程。

与项目织机

如果您正在使用Executors.newVirtualThreadPerTaskExecutor()，这不再是问题（至少在大多数情况下，存在一些例外情况，例如本机代码或在synchronized块中阻塞时 - 在这些情况下，虚拟线程被“固定”到平台线程）。

存在的问题

一个问题是你打破了范式。虽然您的项目是响应式的，但您的某些代码不是响应式的，并且您最终会得到一个代码库，该代码库的某些部分使用响应式代码，而其他部分则不使用响应式代码。但是，如果您正在使用虚拟线程将现有反应式项目迁移到同步代码，并且您计划从项目中删除反应式框架（您可以逐步执行此操作），那么这可能暂时没问题。

但是，请注意，虚拟线程（在撰写本文时）仍然是预览功能，因此可能会对其进行重大更改。因此，您可能还不想迁移到虚拟线程，并等待虚拟线程退出预览，因为切换回来可能非常困难。虚拟线程在 Java 21 中最终确定。

无论如何，不​​要仅仅因为使用了 Loom 的虚拟线程就使用 Reactive 框架启动项目，而在 Reactive 代码中阻塞。相反，要么选择Reactive编程风格，要么使用虚拟线程并同步编写。如果您有应该在平台线程中完成的特定工作（例如 CPU 密集型工作），那么没有什么可以阻止您为此创建平台线程。

毕竟，反应式编程的全部目的不是阻塞线程。如果你想阻止（虚拟）线程，那么使用反应式框架是没有意义的（至少在我看来）。

从反应式迁移到虚拟线程

如何从反应式代码迁移到虚拟线程取决于您和您的项目。如果您的项目已正确模块化，那么选择模块化方法并一个接一个地迁移模块可能是一个好主意。

如果您有使用反应式框架的代码库并且想要切换到虚拟线程，您可以首先配置反应式框架以使用虚拟线程本身Executors.newVirtualThreadPerTaskExecutor()或类似的方法。只要阻塞代码在虚拟线程中运行，这对于自上而下的方法来说不是必需的。

自顶向下

您可以尝试自上而下地迁移它 - 确保调用代码在虚拟线程中执行，并逐步重写以阻止。为此，您可以（因为您位于虚拟线程中）运行一些反应性操作并阻止操作的最终结果。然后，当没有其他反应式代码正在使用该操作时，您可以稍后继续重写反应式操作所调用的反应式代码（调用该操作的所有代码都已重写为虚拟线程）。

自下而上

响应式框架通常提供一种使用另一个线程或类似线程运行非响应式代码的方法。您可以将其配置为使用虚拟线程，然后开始自下而上重写代码。

为了做到这一点，您将开始重写不依赖于其他反应性代码的反应性代码，并更改调用代码以将其视为非反应性代码（同时确保它在虚拟线程中运行）。

混合的

你也可以从中间开始。为此，您确实需要确保您的反应式框架在虚拟线程中运行您的代码。

然后，您将阻止调用的反应性代码，而调用您的代码的代码将重写的代码视为非反应性代码。

但是，您应该注意，这将导致上述打破范式的问题。您可能会遇到半反应半阻塞的代码，处理起来可能很烦人。

模块化

如果您对代码库进行了大量重写，您可能最终会遇到一部分被重写而其他部分没有重写的情况。

如果您的应用程序具有明确分离的模块（可以是微服务，但整体应用程序也可以模块化），您可以从重写一个又一个模块开始。如果您正在使用微服务，则应该能够一次重写一个微服务，而不会影响其他微服务。