尽管 RxJava 在 Android 中会发生 UI 阻塞

Question

我正在使用RxJava将网络访问移动到Android中的单独线程，但我的UI仍然阻塞。

我没有使用错误的可观察量，如下所示：Android RxJava，非阻塞？

以下代码中的代码点 [A]、[B] 和 [C] 按 [A] -> [C] -> [B] 的顺序传递，因此当前线程得到很好的处理，RxJava 在获得结果后调用 [C]。这很好。

此外，与在 UI 线程上进行网络调用相比，阻塞要好得多，但我仍然有轻微的阻塞。调用后，UI 保持流畅，但如果服务器在几毫秒内没有响应，则会阻止。

    private search; // search is an instance variable in the same class
    // [A]
    Observable.just(search.find("something")) // search.find calls the REST endpoint
            .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Search>() {
                @Override public void call(Search search) {
                    // further processing // [B]
                }
            }, new Action1<Throwable>() {
                @Override public void call(Throwable throwable) {
                    // error handler
                }
            });
    // [C]

搜索是 Observable 使用它的同一类中的实例变量，但端点调用是从单独的库执行的，这可能是一个问题吗？应该没关系吧？

我做了什么不该做的坏事吗？

--

查找看起来像这样(为简洁起见，删除了异常处理(：

public Search find(String searchtext) {
    setSearchtext(searchtext);
    SearchEndpoint.find(Session.getUser().getId(), searchtext);
    return this;
}

像这样的搜索端点：

public static Search find(final Long userId, final String searchtext) throws IOException {
    return ApiService.api().searches().find(userId).setFind(searchtext).execute();
}

并调用生成的 Google 云端点库。

Answer 1

试试这个：

Observable.create(new Observable.OnSubscribe<Search>() {
    @Override
    // method signature is from memory - I hope I am correct...
    public void call(Subscriber<? super Search> subscriber) {
        try {
            Search search = search.find("something");
            subscriber.onNext(search);
            subscriber.onCompleted();
        } catch (SomeException e) {
            subscriber.onError(e);
        }
    }
})
// and then continue with your .subscribeOn(...)

澄清一下，也许这使代码的问题更加明显：

Observable.just(search.find("something"))

显然等同于

Search search = search.find("something");
Observable.just(search)

很明显，search.find 是在我们将控制权移交给 rxjava 之前执行的，它是在您当前所在的任何线程上执行的 - 然后从预先计算的值构建一个可观察量并传递值发生在另一个线程上，但这对你没有多大帮助......

Answer 2

我知道

这已经有几个月的历史了 - 但您可以使用 map 运算符运行搜索，而不是create一个全新的 Observable(相对容易出错(：

String search_input = "something"; // this is where you can specify other search terms
Observable.just(search_input) 
      .map(s -> search.find(s))  // search.find calls the REST endpoint
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe( // your subscriber goes here

如果不使用 lambda，则该map函数应如下所示：

.map(new Func1<String, Search>() {
      @Override          
      public Search call(String s) {
          return search.find(s)
      }
 })