Mono：从长时间运行的方法返回结果答案

【问题标题】：Mono: Return results from a long running methodMono：从长时间运行的方法返回结果
【发布时间】：2018-01-05 12:49:35
【问题描述】：

目前我开始使用 Spring + 反应式编程。我的目标是从长时间运行的方法（在数据库上轮询）在 REST 端点中返回结果。我被困在api上。我根本不知道如何在我的FooService.findFoo 方法中将结果返回为Mono。

@RestController
public class FooController {

  @Autowired
  private FooService fooService;

  @GetMapping("/foo/{id}")
  private Mono<ResponseEntity<Foo> findById(@PathVariable String id) {
    return fooService.findFoo(id).map(foo -> ResponseEntity.ok(foo)) //
      .defaultIfEmpty(ResponseEntity.notFound().build())
  }
  ...
}

@Service
public class FooService {

  public Mono<Foo> findFoo(String id) {
    // this is the part where I'm stuck
    // I want to return the results of the pollOnDatabase-method
  }

  private Foo pollOnDatabase(String id) {
    // polling on database for a while
  }
}

【问题讨论】：

这与我的问题有关吗？如果是这样，我不明白。
是的。我的意思是从你的服务类 findFoo 方法，你不能像return Mono.just(pollOnDatabase(id))
谢谢一百万，这就是我想要的。
你使用的是数据库，也就是JDBC，这意味着你基本上是把口红涂在猪身上。 JDBC 是阻塞的且不响应的，因此您基本上一无所获。您可以在不使用异步结果的反应式编程的情况下实现同样的效果。
确实如此。感谢您指出这一点。

标签： spring-mvc spring-boot reactive-programming

【解决方案1】：

使用Mono.fromSupplier 方法！ :)

@Service
public class FooService {

  public Mono<Foo> findFoo(String id) {
    return Mono
      .fromSupplier(() -> pollOnDatabase(id))
      .subscribeOn(Schedulers.boundedElastic());
  }

  private Foo pollOnDatabase(String id) {
    // polling on database for a while
  }
}

使用此方法，我们尽快返回一个 Mono 值，与供应商的恒定时间将由调用者的订阅按需评估。这是调用长时间运行阻塞方法的非阻塞方式。

请注意，如果没有订阅 boundedElastic，阻塞的 pollOnDatabase 方法将阻塞原始线程，从而导致线程饥饿。您可以为每种任务找到不同的时间表here。

请勿在长时间运行的计算中使用 Mono.just，因为它会在返回 Mono 实例之前运行计算，从而阻塞给定线程。

+1：观看this 视频，了解如何避免“反应堆熔毁”。使用一些lib 来检测来自非阻塞线程的阻塞调用。

【讨论】：

【解决方案2】：

这很简单。你可以这样做

@Service
public class FooService {

  public Mono<Foo> findFoo(String id) {
    return Mono.just(pollOnDatabase(id));
  }

  private Foo pollOnDatabase(String id) {
    // polling on database for a while
  }
}

【讨论】：

此解决方案将首先评估pollOnDatabase 函数，然后将结果提供给Mono.just 函数，然后findFoo 函数可以返回。因此，要创建并返回反应式实例，我们将使用长时间运行的数据库调用阻塞主线程。