在 UI 线程上仍可观察到运行

Question

我在运行在新线程上设置我的 RxJava 可观察对象时遇到问题。我正在尝试运行在 Scheduler.io() 上压缩数据库查询，但无论如何它都会阻止 UI 线程。不知道为什么。

数据库代码：

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
        KeyMatchLevel keyMatchLevel, int queryLim) {
    return Observable.just(
            searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim));
}

这是发生压缩的代码：

 public Observable<Set<Airport>> getSearchResultsForAll(String term, int queryLim) {
    List<String> terms = CollectionsUtil.asArrayList(term);
    Observable<List<Airport>> macObservable = onNewThread(
            searchModelsInView(getMacView(), terms,
                    KeyMatchLevel.PREFIXED, queryLim));
    Observable<List<Airport>> nameObservable = onNewThread(
            searchModelsInView(getNameView(), terms,
                    KeyMatchLevel.PREFIXED, queryLim));
    Observable<List<Airport>> codeObservable = onNewThread(
            searchModelsInView(getCodeView(), terms,
                    KeyMatchLevel.PREFIXED, NONE));
    Observable<List<Airport>> regionObservable = onNewThread(
            searchModelsInView(getCityView(), terms,
                    KeyMatchLevel.PREFIXED, NONE));

    ;
    return Observable.zip(macObservable, nameObservable, codeObservable, regionObservable,
            (macList, nameList, codeList, regionList) -> {

                Set<Airport> resultSet = new LinkedHashSet<Airport>();

                Airport mac = macList.get(0);

                if (term.length() > CODE_LEN) {
                    if (lowerCase(mac.getCity()).contains(lowerCase(term))) {
                        handleMACFound(resultSet, mac, regionList);
                        return resultSet;
                    }
                    resultSet.addAll(nameList);
                    resultSet.addAll(regionList);
                    return resultSet;
                }


                boolean macFound = StringUtil.isEquals(lowerCase(term),
                        lowerCase(mac.getCode()));

                if (macFound) {
                    handleMACFound(resultSet, mac, codeList);
                    return resultSet;
                }

                resultSet.addAll(codeList);
                resultSet.addAll(nameList);
                resultSet.addAll(regionList);

                return resultSet;
            });
}

private Observable<List<Airport>> onNewThread(Observable<List<Airport>> observable) {
    return observable.subscribeOn(Schedulers.io());
}

这就是它的名字：

public void searchKey(String searchTerms, Airport... excluding) {
    TBDataBase.getAirportDB()
            .getSearchResultsForAll(searchTerms, MAX_AIRPORT_SEARCHED)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(airports -> {
                setResults(airports, excluding);
                listener.onRegularAirportSearch();
                listener.onSearchPerformed();
            });


}

它似乎应该在 Scheduler.io() 上运行但它阻塞了主线程所以它没有。

Answer 1

看起来问题出在您的 searchModelsTask 中：

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
        KeyMatchLevel keyMatchLevel, int queryLim) {
    return Observable.just(
            searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim));
}

Observable.just 期望您已经拥有该值。所以它甚至在创建可观察对象之前就执行了 searchModelsTask。因此，您尝试将它放在 I/O 调度程序上并不重要。尝试在内部使用延迟：

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
        KeyMatchLevel keyMatchLevel, int queryLim) {
    return Observable.defer(() -> Observable.just(
            searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim)));
}

或

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
            KeyMatchLevel keyMatchLevel, int queryLim) {

return Observable.defer(new Func0<Observable<List<T>>>(){

        @Override
        public Observable<List<T>> call() {
            return Observable.just(
                    searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim));
        }
    });
}

您也可以考虑使用 Observable.callable 而不是 defer。

在 UI 线程上仍可观察到运行

Observable still running on the UI thread

java

android

rx-java

rx-android

在 UI 线程上仍可观察到 运行

Observable still running on the UI thread

java

android

rx-java

rx-android

在 UI 线程上仍可观察到运行