quarkus 反应式叛变线程池管理

如何解决quarkus 反应式叛变线程池管理

背景：我这周才开始使用 Quarkus，尽管我之前曾使用过一些流媒体平台（尤其是 Scala 中的 http4s/fs2）。

使用 quarkus 反应式（使用 mutiny）和任何反应式数据库客户端（mutiny 反应式 postgres、反应式 elasticsearch 等）我有点困惑如何正确管理阻塞调用和线程池。

quarkus 文档建议使用 @Blocking 注释命令式代码或 cpu 密集型代码，以确保将其转移到工作池而不阻塞 IO 池。这是有道理的。

考虑以下事项：

public class Stuff {
  // imperative,cpu intensive
  public static int cpuIntensive(String arg) { /* ... */ }

  // blocking IO
  public static List<Integer> fetchFromDb() { /* ... */ }

  // reactive IO
  public static Multi<String> fetchReactive() { /* ... */ }

  // reactive IO with cpu processing
  public static Multi<String> fetchReactivecpuIntensive(String arg) {
    fetchReactive() // reactive fetch
      .map(fetched -> cpuIntensive(arg + fetched)) // cpu intensive work
  }
}

我不清楚上述每个条件中会发生什么，以及如果它们被从一个没有 @Blocking 注释的 resteasy-reactive rest 端点调用，它们会在何处执行。 >

---

据推测，在没有 @Blocking 的反应式休息端点中使用任何反应式客户端都是安全的。但是在 Uni 中包装阻塞调用对“不安全”clode 来说是否完成了同样的工作？也就是说，返回 Multi/Uni 的任何东西都会在 worker 池中有效运行吗？

（我将打开关于更好地控制线程池的后续帖子，因为我认为没有任何方法可以将反应式 IO 调用“转移”到单独的池而不是 cpu 密集型工作，这将是最佳的。）

编辑

这个问题可能暗示我在询问返回类型（Uni/Multi 与直接对象），但实际上是关于在任何给定时间选择正在使用的线程池的能力。 this mutiny page on imperative-to-reactive 有点实际上回答了我的问题，还有 mutiny infrastructure docs 声明“默认执行程序已配置为使用 Quarkus 工作线程池。”，以及 {{ 3}} 处理我认为的其余部分。

所以我的理解是这样的：

如果我有一个端点，有条件可以返回非阻塞的东西（例如本地非阻塞缓存命中），那么我可以有效地在 IO 线程上以我想要的任何方式返回。但是，如果所述缓存未命中，我可以直接调用反应式客户端或使用 mutiny 在 quarkus 工作池上运行阻塞操作。同样，mutiny 提供了在特定线程池（执行器）上执行任何给定流的控制。

响应式客户端（或在非 IO 池上有效运行的任何东西）可以安全调用，因为 IO 循环只是订阅工作池发出的数据。

最后，我似乎可以将 cpu 绑定池与 IO 绑定工作池分开配置，并将它们作为 executor 明确提供给我需要的任何发射器。所以......我想我现在已经准备好了。

解决方法

这是一个很好的问题！

RESTEasy Reactice 端点的返回类型不会对端点将在哪个线程上提供服务有任何影响。 唯一决定线程的是 @Blocking / @NonBlocking 的存在。

原因很简单：仅仅使用返回类型，无法知道操作是否真的需要很长时间才能完成（从而阻塞线程）。 例如，非反应返回类型并不意味着该操作是 CPU 密集型的（例如，您可能只是返回一些固定的 JSON 响应）。 另一方面，反应类型不能保证操作是非阻塞的，因为正如您提到的，用户可以简单地用反应返回类型包装阻塞操作。