Reactor通量平面图操作员吞吐量/并发控制和实现背压

Question

我正在使用Flux构建我的反应式管道。在管道中，我需要调用3个不同的外部系统REST API，它们对访问速率非常严格。如果我超过每秒速率阈值，我将会受到指数限制。每个系统都有自己的阈值。

我正在使用Spring WebClient进行REST API调用；在3个API中，2个是GET，1个是POST。

在我的反应器管道中，WebClient被包装在平面图中以执行API调用，如下代码所示：

WebClient getApiCall1 = WebClient.builder().build().get("api-system-1").retrieve().bodyToMono(String.class) //actual return DTO is different from string
WebClient getApiCall2 = WebClient.builder().build().get("api-system-2").retrieve().bodyToMono(String.class) //actual return DTO is different from string
WebClient getApiCall3 = WebClient.builder().build().get("api-system-3").retrieve().bodyToMono(String.class) //actual return DTO is different from string

    Flux.generator(generator) // Generator pushes the elements from source 1 at a time

    // make call to 1st API Service
    .flatMap(data -> getApiCall1)
    .map(api1Response -> api1ResponseModified)

    // make call to 2nd API Service
    .flatMap(api1ResponseModified -> getApiCall2)
    .map(api2Response -> api2ResponseModified)

// make call to 3rd API Service
.flatMap(api2ResponseModified -> getApiCall3)
.map(api3Response -> api3ResponseModified)

// rest of the pipeline operators

//end
.subscriber();

问题是，如果我没有将并发性值设置为flatMap，那么在服务启动的几秒钟内，管道执行就会突破阈值。如果我将并发性的值设置为1、2、5、10，那么吞吐量就会变得非常低。

问题是，在不对并发性设置任何值的情况下，如何实现应遵守外部系统速率限制的背压？

Answer 1

Resilience4j支持Reactor的速率限制。请参见：

https://resilience4j.readme.io/docs/ratelimiter

https://resilience4j.readme.io/docs/examples-1#section-decorate-mono-or-flux-with-a-ratelimiter

Answer 2

如果您有“每秒速率”的要求，我将显式地设置流量窗口，并将每个窗口限制在所选的时间段内。这将在不受限制的情况下为您提供最大吞吐量。

我将使用类似于以下内容的助手函数：

public static <T> Flux<T> limitIntervalRate(Flux<T> flux, int ratePerInterval, Duration interval) {
    return flux
            .window(ratePerInterval)
            .zipWith(Flux.interval(Duration.ZERO, interval))
            .flatMap(Tuple2::getT1);
}

它允许您执行：

sourceFlux
        .transform(f -> limitIntervalRate(f, 2, Duration.ofSeconds(1))) //Limit to a rate of 2 per second

然后，您可以根据需要将此映射到您的WebClient调用，同时遵守每个API的现有限制：

sourceFlux
        //...assume API 1 has a limit of 10 calls per second
        .transform(f -> limitIntervalRate(f, 10, Duration.ofSeconds(1)))
        .flatMap(data -> getApiCall1)
        .map(api1Response -> api1ResponseModified)

        //...assume API 2 has a limit of 20 calls per second
        .transform(f -> limitIntervalRate(f, 20, Duration.ofSeconds(1))) 
        .flatMap(api1ResponseModified -> getApiCall2)
        .map(api2Response -> api2ResponseModified)

。。。等等