rx-java - 使用 Resilience4j RateLimiter 减慢 RxJava3 Flowable

Question

我已经尝试过 Resilience4J 示例，但速率限制器似乎不符合我的要求：

@Test
public void simpleReactiveWithRateLimiter() {
    RateLimiterConfig rateLimiterConfig = RateLimiterConfig.custom()
        .limitForPeriod(5)
        .limitRefreshPeriod(Duration.ofSeconds(1))
        .timeoutDuration(Duration.ofMillis(100))
        .build();
    RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.of(rateLimiterConfig);
    RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("tenPerSec", rateLimiterConfig);
    Flowable.interval(1, TimeUnit.MILLISECONDS)
        .takeWhile(f -> f < 100)
        .compose(RateLimiterOperator.of(rateLimiter))
        .blockingSubscribe(i -> log.info("Received {}", i));
}

它仍然在不到 200 毫秒的时间内处理 100 个项目。我一定是做错了什么，不知道是什么。有人可以帮忙吗？

Answer 1

Resilience4j RateLimiter 不限制一个订阅消耗的项目的速率（或一个 Flowable 发出的）。RateLimiter 限制一个 Flowable 上的（并发）订阅数。例如，如果许多客户端想要订阅相同的上游 Flowable，例如 HTTP Call。