通过示例了解 RxJava Zip 运算符

在这篇文章中，我们将通过一个例子来学习 RxJava Zip Operator。

根据 RxJava 官方文档，

Zip 通过指定函数将多个 Observable 的发射组合在一起，并根据该函数的结果为每个组合发射单个项目

Zip 运算符允许我们一次从多个 observables 中获取结果。

假设我们有两个网络 observables，如下所示：

• CricketFansObservable - 一个网络 observable，它返回板球爱好者的用户列表。

• FootballFansObservable - 一个网络 observable，它返回足球迷用户列表。

CricketFansObservable 如下：

fun getCricketFansObservable(): Observable<List<User>> {
    return networkService.getCricketFansObservable()
}

FootballFansObservable 如下：

fun getFootballFansObservable(): Observable<List<User>> {
    return networkService.getFootballFansObservable()
}

我们的 NetworkService 如下：

class NetworkService {

    fun getCricketFansObservable(): Observable<List<User>> {
        return Observable.create<List<User>> { emitter ->
            if (!emitter.isDisposed) { // fetch data from network val data = fetchUserListFromNetwork()
                emitter.onNext(data)
                emitter.onComplete()
            }
        }.subscribeOn(Schedulers.io())
    }

    fun getFootballFansObservable(): Observable<List<User>> {
        return Observable.create<List<User>> { emitter ->
            if (!emitter.isDisposed) { // fetch data from network val data = fetchUserListFromNetwork()
                emitter.onNext(data)
                emitter.onComplete()
            }
        }.subscribeOn(Schedulers.io())
    }

    private fun fetchUserListFromNetwork(): List<User> {
        return listOf()
    }

}

观察者如下：

private fun getObserver(): Observer<List<User>> {
    return object : Observer<List<User>> {
        override fun onSubscribe(d: Disposable) {
            println("onSubscribe")
        }

        override fun onNext(userList: List<User>) {
            println("onNext : $userList")
        }

        override fun onError(e: Throwable) {
            println("onError : ${e.message}")
        }

        override fun onComplete() {
            println("onComplete")
        }
    }
}

过滤喜欢两者的用户的实用功能

private fun filterUserWhoLovesBoth(cricketFans: List<User>,
                                   footballFans: List<User>): List<User> {
    val userWhoLovesBoth = ArrayList<User>()
    for (footballFan in footballFans) {
        if (cricketFans.contains(footballFan)) {
            userWhoLovesBoth.add(footballFan)
        }
    }
    return userWhoLovesBoth
}

让我们一起来了解 zip 运算符。

Observable.zip(
    getCricketFansObservable(),
    getFootballFansObservable(),
    BiFunction<List<User>, List<User>, List<User>> { cricketFans, footballFans -> // here we get both the results at a time. return@BiFunction filterUserWhoLovesBoth(cricketFans, footballFans)
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(getObserver())

首先，我们进行两次网络调用（在使用 Zip Operator 时并行进行），然后在获得结果后，我们正在过滤同时喜欢两者的用户。

通过使用 RxJava Zip 操作符压缩两个 observable，两个网络调用并行运行。当两个都完成时，我们得到两个可观察的结果。这样，我们一次得到两个 observables 的结果。

Zip 运算符的优点：

• 当调度器被正确地提供给每个 observable 时，并行运行所有任务。

• 完成所有任务后，在单个回调中返回所有任务的结果。

这样我们就可以使用 RxJava Zip Operator 来解决有趣的问题。

作者：Amit Shekhar
链接：https://blog.mindorks.com/understanding-rxjava-zip-operator-with-example