Spring Web-Flux 中的背压机制

IT小君 2021-12-01T07:18:04

我是Spring Web-Flux的初学者。我写了一个控制器如下：

@RestController
public class FirstController 
{
    @GetMapping("/first")
    public Mono<String> getAllTweets() 
    {
        return Mono.just("I am First Mono")
    }
}

我知道一种被动的好处是Backpressure，它可以平衡请求或响应率。我想了解如何在Spring Web-Flux 中使用背压机制。

IT小君

WebFlux 中的背压

为了理解 Backpressure 在 WebFlux 框架的当前实现中是如何工作的，我们必须在这里回顾一下默认使用的传输层。我们可能还记得，浏览器和服务器之间的正常通信（服务器到服务器的通信通常也是如此）是通过 TCP 连接完成的。WebFlux 还使用该传输在客户端和服务器之间进行通信。然后，为了理解背压控制术语的含义，我们必须从 Reactive Streams 规范的角度回顾一下背压的含义。

基本语义定义了如何通过背压来调节流元素的传输。

因此，根据该陈述，我们可以得出结论，在 Reactive Streams 中，背压是一种机制，它通过传输（通知）接收者可以消耗多少元素来调节需求；这里我们有一个棘手的问题。TCP 具有字节抽象而不是逻辑元素抽象。我们通常所说的背压控制是指控制向/从网络发送/接收的逻辑元素的数量。尽管TCP有自己的流控制（见意在这里和动画有）这种流量控制仍然是字节，而不是逻辑元素。

在WebFlux模块的当前实现中，背压是由传输流控制来调节的，但是并没有暴露接收者的真实需求。为了最终看到交互流程，请看下图：

为简单起见，上图显示了两个微服务之间的通信，其中左侧发送数据流，右侧使用该流。以下编号列表提供了该图的简要说明：

这是 WebFlux 框架，它适当注意将逻辑元素转换为字节并返回以及将它们传输到 TCP（网络）/从 TCP（网络）接收它们。
这是元素的长时间运行处理的开始，一旦作业完成就会请求下一个元素。
在这里，虽然没有业务逻辑的需求，但 WebFlux 会将来自网络的字节排入队列，而无需他们的确认（业务逻辑没有需求）。
由于 TCP 流量控制的性质，服务 A 仍可能向网络发送数据。

从上图中我们可能会注意到，接收方公开的需求与发送方的需求不同（这里是逻辑元素中的需求）。这意味着两者的需求是隔离的，仅适用于 WebFlux <-> 业务逻辑（服务）交互，并且为服务 A <-> 服务 B 交互暴露较少的背压。所有这一切都意味着 WebFlux 中的背压控制并不像我们预期的那样公平。

所有这一切都意味着 WebFlux 中的背压控制并不像我们预期的那样公平。

但是我还是想知道怎么控制背压

如果我们仍然希望在 WebFlux 中对背压进行不公平的控制，我们可以在 Project Reactor 操作符的支持下做到这一点，例如limitRate(). 以下示例显示了我们如何使用该运算符：

@PostMapping("/tweets")
public Mono<Void> postAllTweets(Flux<Tweet> tweetsFlux) {
    
    return tweetService.process(tweetsFlux.limitRate(10))
                       .then();
}

正如我们从示例中看到的，limitRate()运算符允许定义一次预取的元素数量。这意味着，即使最终订户请求Long.MAX_VALUE元素，limitRate运营商也会将该需求拆分为多个块，并且不允许一次性消耗更多元素。我们可以对元素发送过程做同样的事情：

@GetMapping("/tweets")
public Flux<Tweet> getAllTweets() {
    
    return tweetService.retreiveAll()
                       .limitRate(10);
}

上面的例子表明，即使 WebFlux 一次请求超过 10 个元素，它也会limitRate()将需求限制到预取大小，并防止一次消耗超过指定数量的元素。

另一种选择是实现自己的Subscriber或BaseSubscriber从 Project Reactor扩展。例如，以下是我们如何做到这一点的一个简单示例：

class MyCustomBackpressureSubscriber<T> extends BaseSubscriber<T> {

    int consumed;
    final int limit = 5;

    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        request(limit);
    }
    
    @Override
    protected void hookOnNext(T value) {
        // do business logic there 

        consumed++;
        
        if (consumed == limit) {
            consumed = 0;
            
            request(limit);
        }
    }
}

使用 RSocket 协议公平背压

为了通过网络边界实现逻辑元素背压，我们需要一个合适的协议。幸运的是，有一种叫做Rscoket 协议。RSocket 是一种应用级协议，允许通过网络边界传输实际需求。该协议有一个 RSocket-Java 实现，允许设置 RSocket 服务器。在服务器到服务器通信的情况下，相同的 RSocket-Java 库也提供客户端实现。要了解更多如何使用 RSocket-Java，请在此处查看以下示例。对于浏览器-服务器通信，有一个RSocket-JS实现，它允许通过 WebSocket 连接浏览器和服务器之间的流通信。

基于 RSocket 的已知框架

现在有一些框架建立在 RSocket 协议之上。

变形杆菌

其中一个框架是 Proteus 项目，它提供构建在 RSocket 之上的成熟微服务。此外，变形杆菌属以及与Spring框架集成，因此现在我们可以实现一个公平的背压控制（见的例子有）

进一步阅读

2021-12-01T07:18:04 回复