如何在使用 Observable.zip(...) 时获取 API 响应？

Question

我必须点击 3 API 才能更新同一个屏幕，因此我认为 RxJava 是最快的并行方式。在搜索实现时，我遇到了 Observable.zip(...) 函数，因为它可以并行执行多个 API 命中。

我正在使用 Retrofit 调用 API，并且已经创建了带有 gson 注释的 Pojo class。

示例 Pojo classes:

data class ResponseGetFCData(

    @SerializedName("End")
    val end: String,
    @SerializedName("Uni")
    val uni: String,
    @SerializedName("Y")
    val y: Double
)

data class ResponseAK(

    @SerializedName("End")
    val end: String,
    @SerializedName("Manu")
    val manu: String,
    @SerializedName("Start")
    val start: String,
    @SerializedName("TY")
    val tY: Double

)

示例Api接口：

interface Api{
@GET("GetUniPI")
    fun getFCdata(@Query("pi") pi: String
                 , @Query("uni") uni: String): Observable<ResponseGetFCData>
}

Objective ：根据 3 个 API 中 2 个的响应，我必须计算一些数学计算，第三个 API 响应将携带数据以供回收者查看。在这里，我必须通过从 API 1 中获取 y 并从 API 2 中获取 ty 以及类似的计算来计算 (y * ty)/100。

MyCode: 在 activity onCreate(....):

        val requests = ArrayList<Observable<*>>()
        val backendApi = WinRetrofitHelper.winApiInstance()

        requests.add(backendApi.getFCdata("","","",""))
        requests.add(backendApi.getAKCountry())
        requests.add(backendApi.getRecyclerData("","",""))

        Observable
                .zip(requests) {

                }
                )
                .subscribe({

                    Log.e("Exe Summary","******************Success*******************")

                }) {

                    Log.e("Exe Summary",it.stackTrace.toString())
                }

所以在这里我不知道如何从这 3 个 API 中获取响应以及如何以及在何处计算数学以及我将如何从第 3 个 [=40= 更新 recyclerview 适配器中的数据] 回复。请用更好的方法帮助我理解这一点。

Answer 1

或者你可以试试协程。语法简单易懂

fun toDoWorkConcurrent() {

    job2 = launch {

        try {
            val work1 = async { getThingsDone(43) }
            val work2 = async { getThingsDoneAgain(123) }

            val result = computeResult(work1.await(), work2.await())

            withContext(UI) {
                tvResult1.text = result.toString()
            }

        } catch (exception: Exception) {
            exception.printStackTrace()
        }

    }
}

private fun computeResult(await: Int, await1: Int): Int {
    return await + await1
}

编辑： Source

Answer 2

请这样尝试

 Observable.zip(yourobservalelist, new Function<Object[], Object>() {
        @Override
        public Object apply(Object[] objects) throws Exception {


            return objects;
        }
    })

            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(Disposable disposable) throws Exception {

                }
            })
            .doOnTerminate(new Action() {
                @Override
                public void run() throws Exception {

                }
            })
            .subscribe(new Consumer<Object>() {
                           @Override
                           public void accept(Object o) throws Exception {
                               //Do something on successful completion of allrequests

                               //}
                           }
                       },

                    // Will be triggered if any error during requests will happen
                    new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable e) throws Exception {
                            //Do something on error completion of requests
                            e.printStackTrace();
                        }
                    });


}

Answer 3

尝试使用以下方法：

Observable.zip(
    backendApi.getFCdata("","","",""),
    backendApi.getAKCountry(),
    backendApi.getRecyclerData("","",""),
    Function3<ResponseGetFCData, ResponseAK, List<ResponseMarket>, List<ResponseMarket>> { 
        fcData, akCountry, recyclerData ->

        // Your operation here

        return recyclerData
    })
    .observeOn(AndroidSchedulers.mainThread())
    .doOnSubscribe { /* Loading Start */ }
    .doOnTerminate { /* Loading End */ }
    .subscribe(
            { /* Successfully Synced */ },
            { /* Having error */ }
    )

如何在使用 Observable.zip(...) 时获取 API 响应？

How to fetch the API response while using Observable.zip(...)?

android

rx-java

rx-android