rxJava2 Single.Just() 总是在主线程中执行。如何让它在另一个线程上执行？

Question

在这段代码摘录中，我试图处理一堆数据，但它不能在 UI 线程上，否则体验可能是 ANR。我认为用 rxJava2 很容易做到这一点，但是，数据处理总是在主线程上运行。

数据加载在“演示者”中触发，如下所示：

void loadHistoricalDataFromFile(String filename){
    view.showProgressDialog();
    addDisposable(
            model.loadHistoricalDataObservable(filename)
                    .subscribeOn(rxSchedulers.runOnBackground())
                    .observeOn(rxSchedulers.mainThread())
                    .subscribe(loadedSuccessfully -> {
                        view.hideProgressDialog();
                        if (loadedSuccessfully){
                            view.showSnackBar(R.string.simulator_loaded_data_success, LENGTH_SHORT);
                        } else {
                            view.showSnackBar(R.string.simulator_loaded_data_fail, LENGTH_INDEFINITE);
                        }
                    }));
}

如您所见，我已经使用了 .subscribeOn(rxSchedulers.runOnBackground())

rxSchedulers.runOnBackground() 实现如下：

public class AppRxSchedulers implements RxSchedulers {


    public static Executor backgroundExecutor = Executors.newCachedThreadPool();
    public static Scheduler BACKGROUND_SCHEDULERS = Schedulers.from(backgroundExecutor);
    public static Executor internetExecutor = Executors.newCachedThreadPool();
    public static Scheduler INTERNET_SCHEDULERS = Schedulers.from(internetExecutor);
    public static Executor singleExecutor = Executors.newSingleThreadExecutor();
    public static Scheduler SINGLE_SCHEDULERS = Schedulers.from(singleExecutor);

    @Override
    public Scheduler runOnBackground() {
        return BACKGROUND_SCHEDULERS;
    }

    @Override
    public Scheduler io() {
        return Schedulers.io();
    }

    @Override
    public Scheduler compute() {
        return Schedulers.computation();
    }

    @Override
    public Scheduler mainThread() {
        return AndroidSchedulers.mainThread();
    }

    @Override
    public Scheduler internet() {
        return INTERNET_SCHEDULERS;
    }

    @Override
    public Scheduler single() {
        return SINGLE_SCHEDULERS;
    }
}

Single.Just() 实现如下

Single<Boolean> loadHistoricalDataObservable(String filename){
    return Single.just(loadHistoricalData(filename));
}

private Boolean loadHistoricalData(String filename){
    boolean successful = false;
    String json = FileUtils.readFileAsStringFromExtRam(filename);
    if (json.length() > 0) {
        Gson gson = new Gson();
        historicPriceList = null;
        historicPriceList = gson.fromJson(json, new TypeToken<List<HistoricPrice>>(){}.getType());
        successful = true;
        Timber.d("Successfully loaded file - recreated %d records", historicPriceList.size());
    } else {
        Timber.d("Failed to load file");
    }

    return successful;
}

主要问题是，每当我在其中遇到断点时，loadHistoricalData()我都可以看到它在主线程上运行。它绝对必须在另一个线程上。这怎么可能 ？

Answer 1

The problem is here Single.just(loadHistoricalData(filename));

You are calling the function immediately and then you are passing its result to Single.just(); You need to change it to something like this:

Single.fromCallable(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return loadHistoricalData(filename);
            }
        });

So it will look like this:

Single<Boolean> loadHistoricalDataObservable(String filename){
    return Single.fromCallable(new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    return loadHistoricalData(filename);
                }
            });
}