Можно ли использовать Subject`ы для распространения изменений модели?

Question

Александр Шустанов @alexander-shustanov

Можно ли использовать Subject`ы для распространения изменений модели?

Разрабатываю достаточно тяжелое, высоконагруженное приложение на андроид. Жрет достаточно много памяти, обрабатывает достаточно большое количество данных. Разработка ведется уже больше двух лет, архитектура за это время не раз менялась, в итоге пришли к комбинации flux и rxjava.

В стандартномописании flux ничего про многопоточность не сказано (потому что он для веба придуман), однако все экшены у нас обрабатываются в отдельном потоке (data thread). Это позволяет избежать всякие ConcurrentModificationException и тд., тем более, что некоторые экшены достаточно тяжелые и ui поток просто не потянет. Соответственно, когда вьюхе (view) приходит dataChanged, ей нужно получить данные из некоторого стора. Так как работа с моделью осущетвляется в data thread, то просто store.findSomeEnitityById(id) сделать нельзя. Вместо этого нужно

store.findSomeEnitityByIdObservable(id).observeOn(AndroidSchedulers.mainThread()).subscribe(...)

Представим, что модель часто меняется. DataChanged приходитдостаточно часто и findSomeEnitityByIdObservable создаются чаще, чем успевают отработать старые. Кроме того, в интересующем нас сторе меняется какое-то другое состояние, которое текущую вьюху не интересует. Мне на ум пришел следующий вариант.

В каждом сторе, для каждого минимального кусочка модели, который может понадобится некоторому view создается BehaviorSubject. Например

BehaviorSubject<EntityList> entities = BehaviorSubject.create();

Получение изменений происходит таким образом:

//Метод в сторе
Observable<Entity> findEntityById(long id, long milliseconds) {
    return
        entities.throttleLast(milliseconds, TimeUnits.MILLISECONDS).map(entities -> entities.findById(id));
}
//Вьюшка
void onResume() {
    subscription = store.findEntityById(id, 1000).observeOn(AndroidSchedulers.mainThread()).subscrive(...);
}
void onPause() {
  subscription.unsubscribe();
}

В чем плюс мне кажется - вместо того, чтобы создавать новые Observabe каждый раз, как получаем dataChanged, мы создаем один в onResume, и можем регулировать частоту обновлений. Кроме того, dataChanged говорит нам о том, что состояние стора поменялось, но его состояние может иметь достаточно много аспектов (которые не всегда можно разнести по другим сторам), и dataChanged вовсе не говорит, что поменялось именно то, что нам нужно.

Вопрос в том, правильно ли я делаю, и есть ли другие варианты решения проблемы.