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) 
      } 
     }) 
    } 
} 

虽然我怀疑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闭包也会在我的可观察值上执行。

我已经深入互联网，发现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合同，unscubsccription和backpressure。所以你可以按照你的建议使用RxJavaCallAdapterFactory，或者试试Observable.fromAsync。