使用Rx限制远程资源的使用

Question

我使用RxJava异步处理servlet请求。在每个请求期间，使用flatMap操作符对远程服务API进行大量异步调用。

由于资源限制，我需要限制针对该API的并发请求总数。对于使用其并发参数在单个平面图中调用API的单个Rx流来说，这是微不足道的。但是我的应用程序中有多个独立的流（基本上每个ServletRequest一个），并且每个流都有多个对API的平面图调用。

因此，我认为我必须将所有远程请求汇集到一个执行实际调用的单例流中，然后可以轻松限制并发性。但是，将API响应返回到原始流中似乎并不简单。此外，在这种结构上保持背压似乎很复杂。

另一种选择是使用传统的信号量，但我不确定它的阻塞行为是否适合Rx。

那么，有没有一个既定的模式来实现这样的东西？还是我错过了一个巧妙的操作员组合，完全避免了这些复杂情况？

Answer 1

签出调度程序。when（）。它允许用户从现有调度程序构建修改的调度程序，而不是创建新的线程池。就像使用调度器一样。io（）但限制活动线程的数量。或速率限制调度程序上的onNexts数。计算（）。

Answer 2

在RxJava中，您可以从常规Java执行器创建自己的调度程序：

ExecutorService exec= Executors.newFixedThreadPool(2); //2 Fixed threads
Schedulers.from(exec);

因此，只需为每个资源创建一个具有有限线程数的执行器，并在访问资源时使用该特定调度器。有限的线程数将限制并发调用的数量。

编辑：

显然我误解了这个问题。如果调用是异步的，您可以尝试使用Rx的背压来管理它们。以下是如何使用Rx管理此类呼叫的想法：

您创建了一个“可观察的资源许可”，每当可以调用API时，它都会发出一些东西（某种令牌）。其令牌（许可）创建速率将是该API的最大使用速率。每当某个observable需要调用API时，只需使用许可证observable压缩调用即可。Zip操作符将阻止，直到有可用的许可证，从而将API调用限制在许可证生成的速率上

下面是一个可以观察到的许可证的简单实现，它会发出时间戳：

public class PermitObservable extends Observable<Long> {

    private final long msBetweenEmissions;

    public PermitObservable(long msBetweenEmissions) {
        super(new SyncOnSubscribe<Long, Long>() {
            @Override
            protected Long generateState() {
                return System.currentTimeMillis();
            }

            @Override
            protected Long next(Long state, Observer<? super Long> observer) {
                long nextEmissionAt = state + msBetweenEmissions;
                long timeToWait = nextEmissionAt - System.currentTimeMillis();
                if (timeToWait > 0) {
                    try {
                        Thread.sleep(timeToWait);
                    } catch (InterruptedException e) {
                        observer.onError(e);
                    }
                }
                long now = System.currentTimeMillis();
                observer.onNext(Long.valueOf(now)); // Permit emission
                return now;
            }
        });

        this.msBetweenEmissions = msBetweenEmissions;
    }
}