RxJava并发网络请求

在我以前的博客中，我已经展示了如何使用RxJava 2轻松地将多个网络呼叫连接在一起。但是，以前的方法存在一个缺点，因为它们仍然被顺序调用。

让我们来看看如何实现并发执行.

我们通过mergeWith进行操作

执行速度慢方式

当我们如下使用mergeWith合并两个网络调用时...

val disposable = firstNetworkCall()
    .mergeWith(secondNetworkCall())
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { it -> doSomethingWithIndividualResponse(it) }

所需时间将是firstNetworkCall和secondNetworkCall所需的总时间，如下图所示。

1_6hmjfAEvR7HTzn-uKV9HHQ.png

但是，由于firstNetworkCall和secondNetworkCall彼此不相关，为什么我们不将它们并行化？

优化过的执行方式

为了使其并发请求，对于每个Observable，在其上放置subscibeOn

代码如下:

val disposable = firstNetworkCall().subscribeOn(Schedulers.io())
    .mergeWith(secondNetworkCall().subscribeOn(Schedulers.io())
    // .subscribeOn(Schedulers.io()) // not needed anymore
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { it -> doSomethingWithIndividualResponse(it) }

但是，这看起来有些混乱。我们可以使它更清晰一些，而不是使用mergeWith，而是使用Observable.merge

val disposable = Observable.merge(
        firstNetworkCall().subscribeOn(Schedulers.io()),
        secondNetworkCall().subscribeOn(Schedulers.io()))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { it -> doSomethingWithIndividualResponse(it) }

结果如下所示，所需时间将会是两个请求共同(取请求事件更长的)需要的时间（也许会有一些开销）

1_Hhxlx9kTobjhm-WHW5lbRA.png

通过zipWith进行操作

执行速度慢方式

当我们如下使用zipWith组合两个网络调用时...

val disposable = firstNetworkCall()
    .zipWith(secondNetworkCall(), 
        BiFunction{ 
          firstResonse: ResponseOneType, 
          secondResponse: ResponseTwoType -> 
          combineResult(firstResponse, secondResponse) })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { it -> doSomethingWithCombinedResponse(it) }

所需时间仍然将是firstNetworkCall和secondNetworkCall所需的总时间，如下图所示。

1__hRifuLbCMNoUm2X__f0_A.png

优化过的执行方式

还是同样的, 为了使其并发请求，对于每个Observable，在其上放置一个subscriptionOn

代码如下:

val disposable = firstNetworkCall().subscribeOn(Schedulers.io())
    .zipWith(secondNetworkCall().subscribeOn(Schedulers.io()), 
        BiFunction{ 
          firstResonse: ResponseOneType, 
          secondResponse: ResponseTwoType -> 
          combineResult(firstResponse, secondResponse) })
    // .subscribeOn(Schedulers.io()) // not needed anymore
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { it -> doSomethingWithCombinedResponse(it) }

但是，这看起来有些混乱。我们可以通过使用Observable.zip来使其更加清晰。

val disposable = Observable.zip(
        firstNetworkCall().subscribeOn(Schedulers.io()),
        secondNetworkCall().subscribeOn(Schedulers.io()),
        BiFunction{ 
          firstResonse: ResponseOneType, 
          secondResponse: ResponseTwoType -> 
          combineResult(firstResponse, secondResponse) }))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { it -> doSomethingWithIndividualResponse(it) }

结果如下图所示

1_BItvY8_IM4tdgsXhKACrfg.png

到这里我们就完成了并发的操作

译文链接:
https://proandroiddev.com/rxjava-2-parallel-multiple-network-call-made-easy-1e1f14163eef

最后编辑于：2019.12.04 14:51:35