AndroidObservable 从不在主线程上观察

Question

我正在试验 RxJava for Android .我正在尝试一个简单的示例，我正在尝试卸载后台线程上的较长操作，然后期望返回主线程并执行 ui 操作。

我从 EditText onTextChangeListener 获取搜索文本(通过漂亮的 ButterKnife 注入(inject))，然后将其发送到长时间运行的操作 _searchForContacts( searchText) 返回联系人对象列表。然后，我通过在我的适配器上设置信息并通知数据集更改来继续更新我的 View 。

但是，我在尝试进行任何 ui 更新时遇到异常，因为我从不在主线程上。

这是我的代码:

public class MyTestFragment
    extends Fragment
    implements Observer<List<Contact>> {

    // ...

    private Subscription _searchGuestsSubscription = Subscriptions.empty();

    // ...

    @Override
    public void onDestroyView() {
        super.onDestroyView();
        _searchGuestsSubscription.unsubscribe();
    }

    @Override
    public void onActivityCreated(Bundle savedInstanceState) {
        super.onActivityCreated(savedInstanceState);
        _adapter = new ContactImportAdapter();
        _adapter.setContacts(_searchForContacts(ALL_CONTACTS));
        _listView.setAdapter(_adapter);
    }

    @Override
    public void onNext(List<Contact> contactSearchResults) {
        Timber.d("I'm on the main thread -> " + String.valueOf(Looper.myLooper() == Looper.getMainLooper()));
        _adapter.setContacts(contactSearchResults);
    }

    @Override
    public void onCompleted() {
        _updateView();
    }

    @Override
    public void onError(Throwable e) {
        Timber.e(e, "Oops something went wrong.");
    }

    // ButterKnife ~ onTextChangeListener
    @OnTextChanged(R.id.search_edit_text)
    void onSearchContact(CharSequence searchChars) {
        _searchGuestsSubscription = AndroidObservable.bindFragment(this,
                                                                   _searchGuestsObservable(searchChars.toString()))
                                                     // The below line doesn't seem to have any effect?
                                                     .observeOn(AndroidSchedulers.mainThread())
                                                     .subscribeOn(Schedulers.io())
                                                     .subscribe(this);
    }

    private Observable<List<Contact>> _searchGuestsObservable(final String searchText) {
        return Observable.create(new Observable.OnSubscribe<List<Contact>>() {


            @Override
            public void call(Subscriber<? super List<Contact>> searchResultObserver) {
                // do the search
                List<Contact> contactSearchResults = _searchForContacts(searchText);
                onNext(contactSearchResults);
                onCompleted();
            }
        });
    }

    private void _updateView() {
        _adapter.notifyDataSetChanged();
    }
}

它表示 onNext 总是在后台线程上调用。我得到了通常的 IllegalStateException: The current thread must have a looper! 异常。日志语句还表明我不在 onNext 中的主线程上。

什么给了？

Answer 1

您的问题似乎是因为您直接在 Fragment 中调用 onNext 而不是在订阅者上调用 onNext。

像这样更改您的代码。

private Observable<List<Contact>> _searchGuestsObservable(final String searchText) {
    return Observable.create(new Observable.OnSubscribe<List<Contact>>() {
        @Override
        public void call(Subscriber<? super List<Contact>> searchResultObserver) {
            // do the search
            List<Contact> contactSearchResults = _searchForContacts(searchText);
            searchResultObserver.onNext(contactSearchResults);
            searchResultObserver.onCompleted();
        }
    });
}

当您的 Observable.OnSubscribe<List<Contact>>()由于subscribeOn(Schedulers.io())，您实际上在后台线程上执行了并且您自己调用 Fragment 的 onNext，这就是您获得该异常的原因。