在不同的线程中调用onNext (RxJava Android)

Question

我收到一个对象(城市)，并希望在不同的线程中调度此数据(反映在移动屏幕上并保存到DB此城市)。这是我的代码:

    private CompositeSubscription subscriptions;
    ...
        @Override
            public void onAddButtonClick(String cityName) {
                Observable<City> obs = repository.getCity(cityName);
                if (obs != null){
                    Subscription subscription = obs
                            .subscribeOn(backgroundThread)
                            .observeOn(mainThread)
                            .subscribe(
                                    city -> {
                                        view.cityExists();
                                        repository.saveCityToDb(city);
                                    }
                            );
                    subscriptions.add(subscription);    
                } else {
                    view.showCouldNotFindCity();
                }    
            }
    @Override
    public void subscribe() {
        subscriptions = new CompositeSubscription();
    }
    @Override
    public void unsubscribe() {
        subscriptions.clear();
    }

所以这个方法view.cityExists();我想在UI线程上运行repository.saveCityToDb(city);在后台线程上运行。我该怎么做呢?

方法getCity(...);返回Observable<City>:

public Observable<City> getCity(String name){
        return fileRepository.getCityFromFile(name);
    }

Answer 1

这应该能奏效:

Subscription subscription = obs
       .subscribeOn(backgroundThread)
       .doOnNext(city -> view.cityExists())
       .observeOn(mainThread)
       .subscribe(repository::saveCityToDb);

Answer 2

您可以查找flatMap操作符。它允许与Observable在同一个线程中进行操作(它转换Observable)。

Subscription subscription = obs
   .subscribeOn(backgroundThread)
   .flatMap(repository::saveCityToDb)
   .observeOn(mainThread)
   .subscribe(city -> view.cityExists());

记住，函数

repository::saveCityToDb

必须返回某些内容以使Observable继续。我建议它返回对象Observable.just(city)。