如何在RxAndroid中更改调度程序？

Question

我想在我的项目中使用RxAndroid，我让线程休眠50 my，但是它导致了代码

    public void getTypeAndCommodity() {
    Observable.from(getCommodities())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<Commodity>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(Commodity commodity) {
                }
            });
}

getCommodities：

    private ArrayList<Commodity> getCommodities() {
    // some test info
    ArrayList<Commodity> list = new ArrayList<>();
    for (int i = 0; i < 99; i++) {
        Commodity commodity = new Commodity();
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        commodity.setName("name" + i);
        commodity.setType("type" + (i + 1) / 10);
        list.add(commodity);
    }
    return list;
}

为什么会这样?请帮忙

Answer 1

这是因为getCommodities()是在主线程中执行的，并且只有在io线程中使用subscribeOn(Schedulers.io())执行发射的项。如果您也希望在后台线程中执行getCommidities()，则需要创建一个可观察到的使用create ()方法：

Observable.defer(new Func0<Observable<Object>>() {
        @Override public Observable<Object> call() {
            return Observable.from(getCommodities());
        }
    }).subscribeOn(Schedulers.io())...

如果您需要更多信息：http://blog.danlew.net/2015/07/23/deferring-observable-code-until-subscription-in-rxjava/