使用弹性4j速率限制器减缓RxJava3流量

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发出(的项目速率。RateLimit限制一个Flowable上的(并发(订阅数。例如，如果许多客户端想要订阅相同的上游Flowable，例如HTTP Call。