即使取消订阅观察者，也会调用 Android RxJava onNext

Question

首先我是 Android 新手，但从 iOS 世界开始，我广泛使用 RxSwift，发现了一种我觉得很奇怪的行为。

我将改造调用封装到 RxJava Observable 中（我没有使用 RxJavaCallAdapterFactory，但我真的不认为这是这里的问题）。

fun SearchService.rx_myObjects(query: String?): Observable<List<MyClass>> {
    return Observable.create { observer ->
        val request = getMyObjects(query = query)
        request.enqueue(object : Callback<MyClass>() {
            override fun onSuccess(result: MyClass) {
                Log.v("TEST", "is unsubscribed ${observer.isUnsubscribed} for query $query")
                observer.onNext(result)
                observer.onCompleted()
            }

            override fun onError(error: APIError) {
                observer.onError(error)
            }
        })
    }
}

虽然我怀疑在取消订阅该观察者时会调用 onNext NOT，但情况似乎并非如此

08-04 11:50:39.767 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query par
08-04 11:50:44.264 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed true for query pari
08-04 11:50:44.653 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query par
08-04 11:50:46.358 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query pari
08-04 11:50:47.990 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query pari
08-04 11:50:48.033 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed true for query par

如您所见，最后一个呼叫已取消订阅，并且（您在此处看不到，但它正在发生，因为我有 par 结果而不是 pari 结果）doOnNext 闭包在我的 observable 上执行即使它已取消订阅。

我已经深入互联网并找到了this，这似乎描述了同样的问题，只是我根本没有使用cache。

这是我的订阅电话：

private fun retrieveMyObjects(query: String?) {
        subscription = service.rx_myObjects(query = query)
            .doOnError { showError(it) }
            .doOnNext { this.airports = it }
            .subscribe()
    }

对象调用在 textView 更新时触发

search_view.setOnQueryTextListener(object : SearchView.OnQueryTextListener {
    override fun onQueryTextChange(newText: String?): Boolean {
        subscription?.unsubscribe()
        retrieveMyObjects(newText)

        return true
    }
})

有什么建议吗？我在这里错过了什么吗？

Answer 1

在最坏的情况下：您执行 http 调用。然后，如果文本更改，您取消订阅之前的订阅，然后执行另一个 http 调用。但是，第一个 http 调用的响应刚刚产生。

这就是你得到这种日志命令的原因。由于您已取消订阅 Observable，因此它不应发出新值。为此，您可以在 onNext / oncompleted 块之前添加一个简单的 if ：

request.enqueue(object : Callback<MyClass>() {
        override fun onSuccess(result: MyClass) {

            if(!observer.isUnsubscribed()) {
                  Log.v("TEST", "is unsubscribed ${observer.isUnsubscribed} for query $query")
                  observer.onNext(result)
                  observer.onCompleted()
             }
        }

        override fun onError(error: APIError) {
            if(!observer.isUnsubscribed()) {
                  observer.onError(error)
            }
        }
    })

像这样，您避免发出一个不会被稍后处理的值（当您取消订阅时）

请注意，您正在使用 Observable.create ：您可能会避免使用它，因为您必须管理很多事情，例如 Rx Contract、unscubscription、背压。所以你可以按照你的建议使用RxJavaCallAdapterFactory，或者试试Observable.fromAsync。