java - 尽管使用 RxJava 在另一个线程上订阅，但使用 Google 的 People API 时仍然收到 IllegalStateException

Question

注意:我将 Kotlin 与 RxKotlin 一起使用来实现有用的扩展功能。

我正在尝试使用 Google 的 People API 获取一些用户数据，并且我刚刚将所有 AsyncTasks 和其他内容迁移到使用 RxJava。除了这部分之外，一切正常:

private fun getGooglePerson(service: PeopleService?, account: GoogleSignInAccount) =
    Single.just(service?.people()
                    ?.get("people/me")
                    ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers")
                    ?.execute() ?: Person()) // If null, return an empty person (shouldn't ever be null, though)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeBy(
                    onError = {/*Do something*/},
                    onSuccess = {/*Do something*/}
            )

具体来说，我收到一个 IllegalStateException 消息，从主线程调用此消息可能会导致死锁。我很困惑，因为我觉得调用 subscribeOn 应该将其上方和下方的所有内容(除非 observeOn 跟随它)转移到另一个线程(在本例中为 Schedulers.io())，我尝试像这样重写代码:

private fun getGooglePerson(service: PeopleService?, account: GoogleSignInAccount) =
    service?.people()
                    ?.get("people/me")
                    ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers")
                    ?.execute() ?: Person() // If null, return an empty person (shouldn't ever be null, though)
            .toSingle()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeBy(
                    onError = {/*Do something*/},
                    onSuccess = {/*Do something*/}
            )

仍然没有骰子。我完全被难住了；我应该怎么办？我应该在哪里调用 execute() 以便它在不同的线程上运行？我错过了什么吗？

Answer 1

让我们稍微分解一下。

Single.just(service?.people()
                    ?.get("people/me")
                    ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers")
                    ?.execute() ?: Person())
            .subscribeOn(Schedulers.io())

我们从 Single.just(someValue) 开始。我们可以像这样重写您的代码，它在技术上与您编写的代码相同:

val person = service?.people()
                    ?.get("people/me")
                    ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers")
                    ?.execute() ?: Person())
Single.just(person)

你看到问题了吗？首先执行服务调用，以便为 Single.just 调用提供值。

为了在链的其余部分设置完毕后(即订阅后)运行您的服务调用，您可以使用 defer :

Single.defer(() -> service?.people()
                    ?.get("people/me")
                    ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers")
                    ?.execute() ?: Person())
            .subscribeOn(Schedulers.io())

(语法可能有问题，我没有方便的 IDE)