如何使用 RxJava 发出多个 API 请求并将它们组合起来？答案

【问题标题】：How to make multiple API request with RxJava and combine them?如何使用 RxJava 发出多个 API 请求并将它们组合起来？
【发布时间】：2017-08-01 22:34:17
【问题描述】：

我必须进行 N 次 REST API 调用并结合所有结果，或者如果至少有一个调用失败（返回错误或超时）则失败。我想使用 RxJava，我有一些要求：

能够在某些情况下配置每个单独的 api 调用的重试。我的意思是，如果我有一个 retry = 2 并且我发出 3 个请求，每个请求最多重试 2 次，总共最多 6 个请求。
快速失败！如果一个 API 调用失败了 N 次（其中 N 是重试的配置），剩余的请求是否没有结束也没关系，我想返回一个错误。

如果我希望使用单个线程发出所有请求，我需要一个异步 Http 客户端，不是吗？

谢谢。

【问题讨论】：

标签： asynchronous rx-java observable nonblocking rx-java2

【解决方案1】：

您可以使用Zip 运算符将所有请求在它们结束后压缩在一起，并在那里检查它们是否都成功

 private Scheduler scheduler;
private Scheduler scheduler1;
private Scheduler scheduler2;

/**
 * Since every observable into the zip is created to subscribeOn a different thread, it´s means all of them will run in parallel.
 * By default Rx is not async, only if you explicitly use subscribeOn.
 */
@Test
public void testAsyncZip() {
    scheduler = Schedulers.newThread();
    scheduler1 = Schedulers.newThread();
    scheduler2 = Schedulers.newThread();
    long start = System.currentTimeMillis();
    Observable.zip(obAsyncString(), obAsyncString1(), obAsyncString2(), (s, s2, s3) -> s.concat(s2)
            .concat(s3))
            .subscribe(result -> showResult("Async in:", start, result));
}

private Observable<String> obAsyncString() {
    return Observable.just("Request1")
            .observeOn(scheduler)
            .doOnNext(val -> {
                System.out.println("Thread " + Thread.currentThread()
                        .getName());
            })
            .map(val -> "Hello");
}

private Observable<String> obAsyncString1() {
    return Observable.just("Request2")
            .observeOn(scheduler1)
            .doOnNext(val -> {
                System.out.println("Thread " + Thread.currentThread()
                        .getName());
            })
            .map(val -> " World");
}

private Observable<String> obAsyncString2() {
    return Observable.just("Request3")
            .observeOn(scheduler2)
            .doOnNext(val -> {
                System.out.println("Thread " + Thread.currentThread()
                        .getName());
            })
            .map(val -> "!");
}

在此示例中，我们只是连接结果，但您可以检查结果并在那里执行业务逻辑，而不是这样做。

您也可以考虑merge 或contact。

您可以在这里查看更多示例https://github.com/politrons/reactive

【讨论】：

创建多个Schedulers.newThread() 是不必要的。它为每个创建的工作线程创建一个新线程，因此仅重用相同的调度程序将产生相同的结果。

【解决方案2】：

我建议使用Observable 来包装所有调用。

假设您有调用 API 的函数：

fun restAPIcall(request: Request): Single<HttpResponse>

你想调用这个 n 次。我假设你想用一个值列表来调用它们：

val valuesToSend: List<Request>

Observable
    .fromIterable(valuesToSend)
    .flatMapSingle { valueToSend: Request ->
        restAPIcall(valueToSend)
    }
    .toList() // This converts: Observable<Response> -> Single<List<Response>>
    .map { responses: List<Response> -> 
        // Do something with the responses
    }

因此，您可以从列表的元素中调用 restAPI，并将结果作为列表。

另一个问题是重试。你说你想在达到个人上限时重试。这很棘手。我相信 RxJava 中没有开箱即用的东西。

你可以使用retry(n)，总共可以重试n次，但是那不是你想要的。
还有一个retryWhen { error -> ... }，你可以在那里做给定异常的东西，但你会知道抛出什么元素错误（除非您将元素添加到我认为的异常中）。

我之前没有使用过重试，但是它似乎重试了整个 observable，这并不理想。

我的第一种方法是执行以下操作，将每个元素的计数保存在字典或类似内容中，并且仅在没有单个元素超出限制时重试。这意味着您必须保留一个计数器并在每次超出任何元素时进行搜索。

val counter = valuesToSend.toMap()

yourObservable
    .map { value: String ->
        counter[value] = counter[value]?.let { it + 1 }?: 0 // Update the counter
        value // Return again the value so you can use it later for the api call
    }
    .map { restAPIcall(it) }
    // Found a way to take yourObservable and readd the element if it doesn't exceeds
    // your limit (maybe in an `onErrorResumeNext` or something). Else throw error

【讨论】：