Подождите, пока предыдущий Rx не будет завершен

Question

Привет, я использую RxJava для работы с дисковым хранилищем и выполняю операции. В принципе у меня есть метод, как это:

public Observable<String> getStorageItem(String id, String type) { 
    return Observable.defer(new Func0<Observable<String>>() { 
     // Run db operations to get storage item. 
    } 
} 

Проблема заключается в том, что это возможно, этот метод getStorageItem(...) получает подписку на несколько раз подряд. А операции DB в пределах наблюдаемого не могут выполняться одновременно. Какой у меня лучший вариант? Должен ли я вручную создать некоторую сортировку очереди? Или у RxJava есть какой-то инструмент, который позволяет мне блокировать операцию до тех пор, пока не будет завершена предыдущая?

Answer 1

Вы можете использовать subscribeOn с однопоточными планировщиками, созданных из ExecutorService, чтобы убедиться, что есть только одна операция БД в процессе выполнения:

ExecutorService exec = Schedulers.newSingleThreadExecutor(); 
Scheduler s = Schedulers.from(exec); 

public Observable<String> getStorageItem(String id, String type) { 
    return Observable.fromCallable(() -> { 
     // Do DB operations 
    }); 
} 

getStorageItem("1", "2").subscribeOn(s).subscribe(...); 
getStorageItem("2", "4").subscribeOn(s).subscribe(...); 
getStorageItem("3", "6").subscribeOn(s).subscribe(...); 

Но обратите внимание, что при перемещении вычисления от нити вызывающего абонента, его может выполняться в любое время. Если вам нужно ждать его индивидуально (потому что getStorageItem уже вызван в какой-то поток), вы можете применить toBlocking() после subscribeOn.