在文档中写道,您应该将阻塞代码包装到Mono
中:http://projectreactor.io/docs/core/release/reference/#faq.wrap-blocking
但没有写出如何实际做到这一点。
我有以下代码:
@PostMapping(path = "some-path", consumes = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Mono<Void> doeSomething(@Valid @RequestBody Flux<Something> something) {
something.subscribe(something -> {
// some blocking operation
});
// how to return Mono<Void> here?
}
我在这里遇到的第一个问题是我需要返回一些东西,但我不能。 例如,如果我返回一个Mono.empty
则请求将在通量工作完成之前关闭。
第二个问题是:我如何像文档中建议的那样实际包装阻塞代码:
Mono blockingWrapper = Mono.fromCallable(() -> {
return /* make a remote synchronous call */
});
blockingWrapper = blockingWrapper.subscribeOn(Schedulers.elastic());
不应在控制器处理程序中调用subscribe
,而只需构建一个反应式管道并返回它。最终,HTTP 客户端将请求数据(通过 Spring WebFlux 引擎(,这就是订阅和请求流水线数据的原因。
手动订阅会将请求处理与其他操作分离,这将 1( 删除有关操作顺序的任何保证,以及 2( 如果其他操作使用 HTTP 资源(例如请求正文(,则会中断处理。
在这种情况下,源不是阻塞,而只是转换操作是阻塞。因此,我们最好使用publishOn
来发出信号,表明链的其余部分应该在特定调度程序上执行。如果这里的操作是 I/O 绑定的,那么Schedulers.elastic()
是最好的选择,如果它是 CPU 密集型的,那么Schedulers .paralell
更好。下面是一个示例:
@PostMapping(path = "/some-path", consumes = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Mono<Void> doSomething(@Valid @RequestBody Flux<Something> something) {
return something.collectList()
.publishOn(Schedulers.elastic())
.map(things -> {
return processThings(things);
})
.then();
}
public ProcessingResult processThings(List<Something> things) {
//...
}
有关该主题的更多信息,请查看反应堆文档中的调度程序部分。如果你的应用程序倾向于做很多这样的事情,那么你就会失去反应式流的许多好处,你可以考虑切换到基于 Servlet 的模型,在那里你可以相应地配置线程池。