RxJava 单独过滤一个列表对象内部并返回其对象与过滤后的列表。

Question

我正在使用RxJava2 (Single)和retrofit来处理网络请求，当接收响应时，在对象所包含的其他字段中，有一个对象列表，我试图实现的是过滤掉包含特定 "id "的对象（在列表中），当然我希望它发生在后台线程上，然后在对象的列表被过滤后发出响应。

范围外有没有一种方法可以检测到每个操作者使用的线程？

Answer 1

前提条件

implementation 'io.reactivex.rxjava2:rxjava:2.2.19'
implementation("io.reactivex.rxjava2:rxandroid:2.1.1")

API

data class Result(val id: Int)

data class MyObject(val values: List<Result> = emptyList())

interface RetroFitApi {
    fun getAll(): Single<MyObject>
}

internal class RetroFitApiImpl : RetroFitApi {
    override fun getAll(): Single<MyObject> {
        return Single.fromCallable {
            MyObject(
                listOf(Result(1), Result(2), Result(3))
            )
        }
    }
}

Retrofit的用法，当Retrofit没有自己的线程模型时(androidTest)

import android.os.Looper
import io.reactivex.Single
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.schedulers.Schedulers
import org.assertj.core.api.Assertions.assertThat
import org.junit.Test
import java.util.concurrent.TimeUnit 

@Test
internal fun name() {
    val api = RetroFitApiImpl()

    val test = api.getAll()
        // make sure the subscribe lambda is called in background-thread
        .subscribeOn(Schedulers.io())
        .map { result ->
            // remove all elements, which are id == 1
            result.copy(values = result.values.filterNot { it.id == 1 })
        }
        // move emit, which will probably be emitted from Schedules#io-Thread to Main-Loop. Therefore after applying observeOn the onNext emit in subscribe will be emitted on the UI-Android-Loop
        .observeOn(AndroidSchedulers.mainThread())
        .test()

    test.awaitDone(500, TimeUnit.MILLISECONDS)

    assertThat(test.lastThread()).isEqualTo(Looper.getMainLooper().thread)

    assertThat(test.values()).containsExactly(
        MyObject(values = listOf(Result(2), Result(3)))
    )
}

关于

有什么办法可以检测到每个操作者使用的线程？

你只能用TestConsumer (Single#test)来测试，看最后调用的是哪个Thread。在运行时，你无法知道onNext会在哪个线程上发出，因为RxJava根本不关心线程，只有当你使用observisionOn subscribeOn时才能知道。 默认情况下，onNext会在调用线程上被调用。如果调用的线程是订阅线程，那么如果没有线程参与，你的结果将（可能）被同步发出。

进一步的阅读。

http:/tomstechnicalblog.blogspot.com201602rxjava-understanding-observeon-and.html。