Поддерживает ли rx-java IO Scheduler порядок?

Мне интересно, будет ли Schedulers.io() выполнять задачи в порядке вызывающего абонента.

public class SaveTest {

    @Inject
    private MyRepository myRepository;

    public void save()  {
        Observable.range(0, 20)
                .map(l -> new MyModel(l))
                .observeOn(Schedulers.io())
                .subscribe(myRepository::save);
    }
}

При вставке новых данных, даже если порядок не соблюдается, проблем нет. Но при обновлении ex :

    BankAccount account = new BankAccount();
    account.deposit(500);
    account.withdraw(50);

и каждая операция приводит к сохранению в реактивном потоке, порядок очень важен.

У меня вопрос: поддерживает ли IO Scheduler порядок?


java rx-java2 rx-java
person bubbles    schedule 09.04.2020    source источник

Ответы (1)


arrow_upward
4
arrow_downward

Да, реактивный поток поддерживает порядок событий, даже если вы передаете события другому планировщику, подобному этому.

Внутри события помещаются в очередь, и планировщик выбирает однопоточного рабочего процесса (как описано здесь) выбирает их из очереди по порядку.

person Joni    schedule 09.04.2020
comment
поэтому каждый раз, когда мы вызываем .observeOn(Schedulers.io()), он просматривает поток из пула, и этот поток будет принимать события в порядке их выбросов? - person bubbles; 09.04.2020
comment
В основном да. Технически поток выбирается при подписке, а не при вызове observeOn. - person Joni; 09.04.2020
comment
Верно ли это для других планировщиков? - person bubbles; 09.04.2020
comment
Все встроенные планировщики документируют, что они используют однопоточные рабочие процессы, но вы можете создать собственный планировщик с многопоточными рабочими процессами с помощью Schedulers.from — документация предупреждает вас об этом: Однако обратите внимание, если предоставленный экземпляр ScheduledExecutorService не является однопоточным, задачи, запланированные с временной задержкой, близкой друг к другу, могут в конечном итоге выполняться в другом порядке, чем исходный вызов schedule(). - person Joni; 09.04.2020