RxJava-AndroidRxJava onNext即使在观察者取消订阅时也会调用

Question

首先，我是安卓系统的新手，但从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)
            }
        })
    }
}

虽然我怀疑当观测者取消订阅时会调用onNextNOT，但的情况似乎并非如此

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上执行，即使它被取消订阅。

我在互联网上搜索了一下，发现这似乎描述了同样的问题，只是我根本没有使用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：您可能会避免使用它，因为您必须管理很多事情，如处方合同、取消订阅、背压。因此，您可以按照建议使用RxJavaCallAdapterFactory，或者尝试使用Observable.fromAsync。