如何使用 Retrofit 2.0 限制并行请求？答案

【问题标题】：How to limit parallel requests with Retrofit 2.0?如何使用 Retrofit 2.0 限制并行请求？
【发布时间】：2016-05-18 15:00:37
【问题描述】：

我的目标是通过抛出异常来限制并行执行请求的数量。

例如，我只想要一个执行请求：

someApi.getUser(
    result -> print("ok: " + result), exception -> print("error: " + exception)
); // this request will be executed in 5 seconds

someApi.getServerInfo(
    result -> print("ok: " + result), exception -> print("error: " + exception)
); // there I want to catch exception like ExecutorIsBusy

如何使用 Retrofit 2.0 实现它？

【问题讨论】：

这并不能完全回答您的问题，但它可能会为您指明正确的方向。我在这里考虑两种可能性。首先，查看此链接stackoverflow.com/questions/23255523/…。它适用于 Retrofit 1 而不是 2，所以我不确定它不能应用。其次，您可以将调用转换为可观察的并使用条件运算符来防止其他订阅发生，直到当前订阅完成。
速率限制最好在服务器端设置...根据您使用的框架（如 laravel、django、express 等）使用用户 ID 或 IP 地址。这种方式不会影响性能，而且仅那些恶作剧的用户将面临速率限制

标签： android retrofit2

【解决方案1】：

我不确定抛出Exception 是不是最好的方法，但我不知道你的用例所以我不会讨论这一点:)

不管怎样，@Daniel 的评论其实指向了一个不错的方向。如果您将retrofit 与OkHttp 一起使用，则OkHttpClient 将处理“并发请求”的内容。阅读文档，您可以看到 OkHttp 使用 Dispatcher 来处理并行异步请求 (Dispatcher docs)。

所以两个有趣的点是：

方法setMaxRequests(int maxRequests)：定义最大并发请求数
方法executed(RealCall call)：实际执行一个请求

我认为你可以这样做来实现你的目标：

创建自定义Dispatcher 类
覆盖executed(RealCall call) 方法，如果当前请求的数量高于maxRequests，则抛出异常
在OkHttpClient 中使用自定义Dispatcher 和retrofit

【讨论】：

谢谢。我会试试你的解决方案。
> I am not sure that throwing an Exception is the best way to do it; 我只需要同时执行一个请求。我需要在当前执行之前拒绝任何请求。还有什么比抛出异常更合适的呢？ @Daniel 建议使用 Observable，但由于我刚开始使用响应式编程，所以我无法想象该怎么做。
我也不知道 ;) 如果你不应该尝试你的请求而不是“尝试和失败” ;)
如果您要使用RxJava，您可以在observable 运行时将boolean 值设置为true，一旦observable 完成，您可以将标志设置回到false 在onCompleted()。然后，任何时候您运行 observable 并且标志已设置为 true，您都不会拨打电话。
我刚刚尝试了这种方法，它在我的示例项目中有效，因此它可能对您有用。让我知道是否可以提供任何进一步的帮助。

【解决方案2】：

使用 RxJava & 采用上述 cmets 中的方法，这里是一个例子：

假设这些操作附加到按钮上。

public void runObservable1(View view) {
    if (!taskRunning){
        try{
            taskRunning = true;
            subsonicService.runTask1()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .compose(this.<Result>bindUntilEvent(ActivityEvent.DESTROY))
                    .subscribe(new Subscriber<Result>() {
                        @Override
                        public void onCompleted() {
                            taskRunning = false;
                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onNext(Result result) {
                            //your logic here
                        }
                    });
        } catch (IOException e) {
            e.printStackTrace();
        }
    } else {
        Toast.makeText(this, "Task is running you must wait", Toast.LENGTH_SHORT).show();
    }


}

public void runObservable2(View view) {

    if (!taskRunning){
        try{
            taskRunning = true;
            subsonicService.runTask2()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .compose(this.<Result>bindUntilEvent(ActivityEvent.DESTROY))
                    .subscribe(new Subscriber<Result>() {
                        @Override
                        public void onCompleted() {
                            taskRunning = false;
                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onNext(Result result) {
                            //Logic here
                        }
                    });
        } catch (IOException e) {
            e.printStackTrace();
        }
    } else {
        Toast.makeText(this, "Task is running you must wait", Toast.LENGTH_SHORT).show();
    }
}

我也不是 RxJava 方面的专家，所以可能有一个操作员可以让这变得更容易。

【讨论】：