返回磁通量和背压的 Web 磁通控制器

Question

在Spring WebFlux中，我有一个类似于这样的控制器：

@RestController
@RequestMapping("/data")
public class DataController {

  @GetMapping(produces = MediaType.APPLICATION_JSON_VALUE)
  public Flux<Data> getData() {
    return <data from database using reactive driver>
  } 
}

1. 订阅出版商到底是什么
2. 什么（如果有的话）提供了背压

对于上下文，我试图评估在这种特定情况下使用Spring WebFlux是否比Spring MVC有优势。

Answer 1

订阅发布者到底是什么？

框架（在本例中为Spring）

一般来说，你不应该在你自己的应用程序中订阅——框架应该在必要的时候订阅你的发布者。在spring的上下文中，每当一个相关的请求到达控制器时。

什么(如果有的话)提供背压？

在这种情况下，它只受连接速度的限制（我相信Webflux会查看底层TCP层），然后根据需要请求数据。然而，上游流量是否会听到背压则是另一回事——它可能会这样做，或者它可能只是向消费者提供尽可能多的数据。

对于上下文，我试图评估在这种特定情况下使用Spring WebFlux是否比Spring MVC有优势。

主要优势是能够仅用几个线程保持大量连接打开——因此没有上下文切换的开销。（这不是唯一的优势，但大多数优势通常归结为这一点。）通常，如果您需要同时保持数千个连接打开，这只是一个值得考虑的优势。

主要的缺点是，反应式代码看起来与标准Java代码非常不同，因此必然更加复杂。调试也很困难——例如，普通堆栈跟踪几乎毫无用处（尽管它们是工具）

Answer 2

注意：我不是Spring框架的开发者，所以欢迎任何评论。

订阅发布者到底是什么？

它是对端口（服务器初始化本身）的长期订阅。因此，ReactorHttpServer.class具有以下方法：

@Override
protected void startInternal() {
    DisposableServer server = this.reactorServer.handle(this.reactorHandler).bind().block();
    setPort(((InetSocketAddress) server.address()).getPort());
    this.serverRef.set(server);
}

订阅者是绑定方法，据我所知，请求（Long.MAX_VALUE），因此这里没有背压管理。

请求处理的重要部分是方法< code > handle(this . reactor handler)。< code>reactorHandler是< code > ReactorHttpHandlerAdapter 的一个实例。堆栈的更高层(在< code > ReactorHttpHandlerAdapter 的< code>apply方法中)是< code > dispatcher handler . class 。这个类的java文档以“HTTP请求处理器/html" target="_blank">控制器的中央调度程序”开始。分派给已注册的处理程序来处理请求，提供了方便的映射工具。。它有中心方法:

@Override
public Mono<Void> handle(ServerWebExchange exchange) {
    if (this.handlerMappings == null) {
        return createNotFoundError();
    }
    return Flux.fromIterable(this.handlerMappings)
            .concatMap(mapping -> mapping.getHandler(exchange))
            .next()
            .switchIfEmpty(createNotFoundError())
            .flatMap(handler -> invokeHandler(exchange, handler))
            .flatMap(result -> handleResult(exchange, result));
}

在这里，实际的请求处理发生了。响应写入handleResult中。现在取决于实际的服务器实现，以及结果的编写方式。

对于默认服务器，即Reactor Netty，它将是< code > reactorserverhttpresponse . class 。这里可以看到方法< code>writeWithInternal。这个函数获取处理程序方法的< code>publisher结果，并将其写入底层HTTP连接:

@Override
protected Mono<Void> writeWithInternal(Publisher<? extends DataBuffer> publisher) {
    return this.response.send(toByteBufs(publisher)).then();
}   

NettyOutbound的一个实现。send（…）isreactor.netty.channel.ChannelOperations。对于流量返回的特定情况，此实现在＜code＞MonoSendMany.class＜code＞中管理NIO。这个类使用＜code＞sendmanInner订阅（…）。类，它通过实现＜code＞订户

所以，

什么(如果有的话)提供背压？

...是的，提供了背压，例如通过SendManyInner.class，但是也存在其他实现。

对于上下文，我试图评估在这种特定情况下使用Spring WebFlux是否比Spring MVC有优势。

我认为，它绝对值得评估。然而，对于性能，我想，结果将取决于并发请求的数量，也可能取决于您的Data类的类型。一般来说，Web通量通常是高吞吐量、低延迟情况的首选，我们通常会在我们的环境中看到更好的硬件利用率。假设您从数据库中获取数据，您可能会使用也支持反应式的数据库驱动程序获得最佳结果。除了性能之外，背压管理始终是查看Web通量的一个很好的理由。自从我们采用Webflow以来，我们的数据平台再也没有稳定性问题了（不要说，没有其他方法可以拥有稳定的系统，但在这里许多问题都可以开箱即用地解决）。

作为附带说明：我建议仔细查看Schedulers，我们最近通过为慢速DB访问选择正确的cpu时间获得了30%的cpu时间。

编辑:在https://docs . spring . io/spring/docs/current/spring-framework-reference/we b-reactive . html # web flux-fn-handler-functions中，参考文档明确指出:

ServerRequest和ServerACK是不可变的接口，提供对HTTP请求和响应的JDK 8友好访问。请求和响应都为主体流提供Reactive Streams背压。