Загрузка из сети, отображение в пользовательском интерфейсе и сохранение на диск с помощью RxJava

Question

Теперь я работаю над задачей, которая будет извлекать данные из сети, отображаться в пользовательском интерфейсе и сохранять данные в локальной базе данных для использования позже. В настоящее время я делаю следующим образом:

//Cloud loading 
Observable<MyModel> loadDataFromCloud(); 

//Database 
Observable<MyModel> saveDataToDisk(MyModel myModel); 

//Load data 
loadDataFromCloud() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .flatMap(new Func1<MyModel, Observable<MyModel>>() { 
    @Override 
    public Observable<MyModel> call(MyModel myModel) { 
     return saveDataToDisk(myModel); 
    } 
    }) 
    .subscribe(new Subscriber<MyModel>() { 
    @Override 
    public void onCompleted() { 

    } 

    @Override 
    public void onError(Throwable e) { 
     //exception handling 
    } 

    @Override 
    public void onNext(MyModel myModel) { 
     //bind myModel to UI 
    } 
    }); 

Этот подход хорошо работает, но так как модель данных становится более сложной, и база данных становится больше, это не эффективно больше, потому что все транзакции базы данных теперь выполняются на основной thread И пользовательский интерфейс должен ждать, пока транзакции не будут выполнены, чтобы иметь данные.

Я сейчас меняется к этому новому подходу:

//A Singleton 
Scheduler scheduler = Schedulers.from(Executors.newSingleThreadExecutor()); 

//Load data 
loadDataFromCloud() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .flatMap(new Func1<MyModel, Observable<MyModel>>() { 
    @Override 
    public Observable<MyModel> call(MyModel myModel) { 
     try { 
     //bind myModel to UI 
     return Observable.just(s); 
     } catch(Exception e) { 
     return Observable.error(e) 
     } 
    } 
    }) 
    .observeOn(scheduler) 
    .flatMap(new Func1<MyModel, Observable<MyModel>>() { 
    @Override 
    public Observable<MyModel> call(MyModel myModel) { 
     return saveDataToDisk(myModel); 
    } 
    }) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Subscriber<MyModel>() { 
    @Override 
    public void onCompleted() { 

    } 

    @Override 
    public void onError(Throwable e) { 
     //exception handling 
    } 

    @Override 
    public void onNext(MyModel myModel) { 
     //do nothing 
    } 
    }); 

Я использую Schedulers.from(Executors.newSingleThreadExecutor()) чтобы убедиться, что все транзакции базы данных будет потокобезопасной.

И у меня есть несколько вопросов:

1. ли с помощью Schedulers.from(Executors.newSingleThreadExecutor()) чтобы убедиться, что все транзакции базы данных будет поточно-это правильный путь?
2. Есть ли лучший способ для достижения этой цели, так как последний onNext() просто делать ничего, кажется, неразумно
3. С RxJava мы можем начать делать две задачи, как то же время на двух разных ниток (показывая на пользовательском интерфейсе и сохранении диск)

Я искал какое-то время и получил некоторые статьи, такие как this one, но они не идут непосредственно на мою проблему.

Answer 1

Во-первых, вы в первую очередь выполняете операцию ввода-вывода в основной теме, это плохая практика, и в андроиде это может привести к анр.

1. В основном это должно синхронизировать, но до тех пор, как вы используете один и тот же Scheduler через приложение, в противном случае вы будете создавать новый поток для каждого подписаться на этой наблюдаемой. Но, некоторые немного странные, чтобы сделать это, так как операции DB обычно являются потокобезопасными.
2. В вашем сценарии вы хотите показать ui, а вы можете saveDataToDisk() параллельно, не дожидаясь его, так что вы можете использовать оператор doOnNext для запуска операции сохранения на другом наблюдаемом (вы не заботитесь о его результатах) без наблюдая его результат.

Так, чтобы подвести итог, вы должны что-то вроде:

loadDataFromCloud() 
    .doOnNext(data - > saveDataToDisk().subscribeOn(Schedulers.io())) 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(data -> //uodate your ui) 

Dan Лью пост является отличным примером, вы должны следовать примеру там.