在 Spring Webflux Annotated 控制器中的不同线程中运行任务答案

【问题标题】：Running Tasks in different thread in Spring Webflux Annotated controller在 Spring Webflux Annotated 控制器中的不同线程中运行任务
【发布时间】：2019-12-17 11:05:45
【问题描述】：

我有一个 Spring Webflux Annotated 控制器，如下所示，

 @RestController
public class TestBlockingController {

  Logger log = LoggerFactory.getLogger(this.getClass().getName());

  @GetMapping()
  public Mono<String> blockForXSeconds(@RequestParam("block-seconds") Integer blockSeconds) {
    return getStringMono();
  }

  private Mono<String> getStringMono() {
    Integer blockSeconds = 5;
    String type = new String();
    try {
      if (blockSeconds % 2 == 0) {
        Thread.sleep(blockSeconds * 1000);
        type = "EVEN";
      } else {
        Thread.sleep(blockSeconds * 1000);
        type = "ODD";
      }
    } catch (Exception e) {
      log.info("Got Exception");
    }
    log.info("Type of block-seconds: " + blockSeconds);
    return Mono.just(type);
  }
}

如何让 getStringMono 在与 Netty 服务器线程不同的线程中运行。我面临的问题是，当我在服务器线程中运行时，吞吐量基本上会减少（每秒 2 个请求）。如何在单独的线程中运行 getStringMono。

【问题讨论】：

标签： spring-boot spring-webflux project-reactor

【解决方案1】：

您可以使用subscribeOn 运算符将任务委托给不同的线程池：

Mono.defer(() -> getStringMono()).subscribeOn(Schedulers.elastic());

不过，您必须注意，在响应式应用程序中应不惜一切代价避免这种类型的阻塞。如果可能，请使用支持非阻塞 IO 并返回承诺类型（Mono、CompletableFuture 等）的客户端。如果您只是想人为延迟，请改用Mono.delay。

【讨论】：

谢谢，您的回答有帮助。

【解决方案2】：

你可以使用 Mono.defer() 方法。方法签名为：

public static <T> Mono<T> defer(Supplier<? extends Mono<? extends T>> supplier)

您的 Rest API 应该如下所示。

@GetMapping()
public Mono<String> blockForXSeconds(@RequestParam("block-seconds") Integer blockSeconds) {
    return Mono.defer(() -> getStringMono());
}

defer 操作符可以让这个源变得懒惰，每次有新订阅者时重新评估 lambda 的内容。这将增加您的 API 吞吐量。

Here可以查看详细分析。

【讨论】：

谢谢，您的回答有帮助。
如果没有明确指定调度程序，此代码将继续在同一线程上工作。 subsribeOn/publishOn 是必填项。
谢谢先生。如果我错了，请继续建议我。这肯定会增加我的知识。