使用RxJava对多线程创建的多个可观察对象进行速率限制

Question

我正在开发一个简单的REST应用程序，它利用RxJava向远程服务器发送请求（1）。对于REST API的每个传入请求，都会向（1）发送一个请求（使用RxJava和RxNetty）。一切正常，但现在我有了一个新的用例：

为了不让太多的请求轰炸（1），我需要实施速率限制。解决这个问题的一种方法（我假设）是将在向（1）发送请求时创建的每个可观察的（2）添加到另一个执行实际速率限制的（2）中。（2） 然后将或多或少地像队列一样，以尽可能快的速度处理出站请求（但不超过速率限制）。下面是一些伪代码：

Observable<MyResponse> r1 = createRequestToExternalServer() // In thread 1
Observable<MyResponse> r2 = createRequestToExternalServer() // In thread 2

// Somehow send r1 and r2 to the "rate limiter" observable, (2)

rateLimiterObservable.sample(1 / rate, TimeUnit.MILLISECONDS)

我将如何使用Rx/RxJava来解决这个问题？

Answer 1

我会使用一个热定时器和一个原子计数器，在给定的持续时间内跟踪剩余的连接：

int rate = 5;
long interval = 1000;

AtomicInteger remaining = new AtomicInteger(rate);

ConnectableObservable<Long> timer = Observable
        .interval(interval, TimeUnit.MILLISECONDS)
        .doOnNext(e -> remaining.set(rate))
        .publish();

timer.connect();

Observable<Integer> networkCall = Observable.just(1).delay(150, TimeUnit.MILLISECONDS);

Observable<Integer> limitedNetworkCall = Observable
        .defer(() -> {
            if (remaining.getAndDecrement() != 0) {
                return networkCall;
            }
            return Observable.error(new RuntimeException("Rate exceeded"));
        });

Observable.interval(100, TimeUnit.MILLISECONDS)
.flatMap(t -> limitedNetworkCall.onErrorReturn(e -> -1))
.take(20)
.toBlocking()
.forEach(System.out::println);