project-reactor - 在 JAX-RS 驱动的服务器中使用 react 器时如何处理阻塞调用？

Question

为了处理 HTTP 请求，我们必须进行阻塞调用(例如 JDBC 调用)作为 Mono/Flux 的一部分基于的过程。我们目前的计划是这样的:

// I renamed getSomething to processJaxrsHttpRequest
CompletionStage<String> processJaxrsHttpRequest(String input) {
  return Mono.just(input)
      .map(in -> process(in))
      .flatMap(str -> Mono.fromCallable(() -> jdbcCall(str)).subscribeOn(fixedScheduler))
      .flatMap(str -> asyncHttpCall(str))
      .flatMap(str -> Mono.fromCallable(() -> jdbcCall(str)).subscribeOn(fixedScheduler))
      .toFuture();
}

哪里 fixedScheduler跨 HTTP 请求并发使用。

我们希望得到一些关于在相当数量的通量内处理块调用的策略的反馈。当然，我们明白，如果我们所有的请求都通过这些阻塞调用流过，那么我们最好不要使用 react 器(在公认的好处理 API 之外)。

更新:谢谢 bsideup为 this answer .但是，我应该对我的问题更具体一点。

我的总体问题是如何有效地跨多个通量使用阻塞调用，如果这些通量可以被大量创建/订阅。我们试过 the suggested approach ，但它会导致线程爆炸和快速 OOM。所以我们正在考虑使用共享调度程序。所以..这是我的问题。

在我描述的情况下，您是否建议使用共享调度程序( fixedScheduler )？如果没有，你会指出我的任何方向吗？

如果使用共享调度器很好，这是否是它的一个很好的实现:Schedulers.newParallel("blocking-scheduler", maxNumThreads) ?

更新 2:刚刚在 Schedulers#newParallel 上挖了一个大坑并意识到这是行不通的，因为它“拒绝”了阻塞调用。

真的很感激任何提示!

Answer 1

虽然 subscribeOn确实是处理阻塞调用的一种方式，您的用法还可以，您也可以使用 publishOn .
它将处理移动到提供的 Scheduler , 除非其他 publishOn指定:

CompletionStage<String> getSomething(String input) {
  return Mono.just(input)
      .map(in -> process(in)) // process must be non-blocking, or go after publishOn
      .publishOn(Schedulers.boundedElastic())
      .map(::jdbcCall)
      .flatMap(str -> asyncHttpCall(str))
      .publishOn(Schedulers.boundedElastic())
      .map(::jdbcCall)
      .toFuture();
}

如您所见，您也可以继续使用异步调用。只要确保你没有阻塞非阻塞调度程序(在那个例子中，我在 publishOn 之后再次使用 flatMap 因为 asyncHttpCall 可能从非阻塞调度程序完成)