в RxJava что считается асинхронной границей?

Question

В настоящее время я глубоко погружаюсь в опасности потоков Java и, в частности, как они применяются к RxJava. Ответы приветствуются, но для предупреждения неизбежного RTFM я также буду благодарен за ссылки на упомянутое руководство.

Я много слышал термин Асинхронная граница, и у меня есть общее представление, но я не нашел строгого определения.

Где в этих примерах асинхронные границы?

// example 1
Observable.just(123)
         .map(i -> i + 1)
         .subscribe(System.out::println); 

// example 2
Observable.just(123)
         .observeOn(Schedulers.newThread())
         .map(i -> i + 1)
         .subscribe(System.out::println); 

// example 3
Observable.just(123)
          .subscribeOn(Schedulers.newThread())
          .map(i -> i + 1)
          .subscribe(System.out::println);

Существуют ли другие примеры, которые имеют отношение к делу.

Answer 1

Оператор является асинхронной границей, если 1) он использует время, 2) принимает асинхронный планировщик и / или 3) связан с более чем одним реактивным источником.

Для 1) это может означать, что нисходящий поток может принимать сигналы в исходном потоке или стандартном планировщике по умолчанию. Пример: timeout.

Для 2) оператор принимает асинхронный планировщик и направляет события и / или управляющие сигналы через этот планировщик. Стандартные планировщики, кроме батута, являются асинхронными планировщиками. Пример: observeOn, subscribeOn, unsubscribeOn.

Для 3) задействовано несколько источников, каждый из которых может излучать из своего собственного потока. В этом случае исход и последующий поток могут быть уведомлены в любом из этих потоков (или даже в потоке, инициирующем первоначальную подписку), и часто все ставки не соответствуют тому, каким потоком это будет. Пример: merge, flatMap.

Помимо этого, если задействован Flowable, обратное давление внутри операторов может действовать как асинхронная граница, так как излучение из восходящего потока должно координироваться с запросом из нисходящего / произвольного потока. В этом случае любой из участвующих потоков может быть ответственным за излучение сигнала, и он может изменяться динамически. Пример: onBackpressureBuffer, observeOn, create