如何使用 RXJava 多次调用方法 - 我使用哪个运算符？

Question

我有一个通过网络调用来检索数据的函数，我需要调用它 5 次。我的方法：

@Throws(IOException::class)
private fun getWeather(future : Int): String {
    var twitterURL = Constants.API_URL
    if (future > 0) {
        twitterURL += "future$future.json"
    }
        val urlBuilder = HttpUrl.parse(Constants.API_URL).newBuilder()
        val url = urlBuilder.build().toString()

        val request = Request.Builder()
                .url(url)
                .build()
        val client = OkHttpClient()
        val response = client.newCall(request).execute()
        val body = response.body()
        return if (response.code() == HttpURLConnection.HTTP_OK) {

            body.string()
        } else {

            throw IOException("Bad Request: Server Response" + response.code().toString() + " " + response.message())
        }


}

我想使用 observable 模型做类似的事情：

private val array = DoubleArray(5)
Observable.fromArray(listOf(1, 2, 3, 4, 5))
      .subscribeOn(Schedulers.newThread()) 
      .observeOn(AndroidSchedulers.mainThread())) 
      //observation on the main thread
      .subscribe(object:Subscriber<Int>(){
        override fun onCompleted() {
            calculateStandardDeviation(array)
        }

        override fun onError(e: Throwable?) {
         //TODO : Handle error here
        }

        override fun onNext(t: Int?) {
         val string = getWeather(t)
         val gson = Gson()
         val weather = gson.fromJson(string,TwitterWeather::class.java)
         array[t-1] = weather.temperature

        }
       })

但实际上，onNext() 在主线程上运行。我想在 Schedulers.io() 线程上运行它。

Answer 1

.observeOn(AndroidSchedulers.mainThread()) 告诉订阅回调将在哪个调度程序上运行，包括onNext 运行。所以就用.observeOn(Schedulers.io())

查看这篇文章了解更多详情： https://medium.com/upday-devs/rxjava-subscribeon-vs-observeon-9af518ded53a

从概念上讲，您需要将网络调用从订阅逻辑中移出并使其可观察，如下所示：

Observable.fromIterable(listOf(1, 2, 3, 4, 5))
            .flatMap { t ->
                Observable.fromCallable {
                    val string = getWeather(t)
                    val gson = Gson()
                    val weather = gson.fromJson(string, TwitterWeather::class.java)
                    weather.temperature
                }.subscribeOn(Schedulers.io())
            }.toList()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({ temperatures ->
                calculateStandardDeviation(temperatures)
            }, {
                //TODO : Handle error here
            })