使用 Resilience4j RateLimiter 减慢 RxJava3 Flowable

问题描述

我已经尝试了 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 个项目。我一定是做错了什么，不知道是什么。有人可以帮忙吗？

解决方法

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