Порядок установки Планировщики в Rx

Question

Я действительно новым для Rx, просто интересно, как же порядок subscribeOn воздействия Observable

//This will not print anything 
Observable.just("whatever") 
     .flatMap { s -> Observable.just(s.length) } 
     .subscribeOn(Schedulers.newThread()) 
     .subscribe(::println) 

//This prints the length 
Observable.just("whatever") 
     .subscribeOn(Schedulers.newThread()) 
     .flatMap { s -> Observable.just(s.length) } 
     .subscribe(::println) 

Что случилось под капотом и почему?

Answer 1

Планировщики RxJava используют потоки демона, которые могут останавливаться или вообще не запускаться, если «основной» поток Java завершает работу. Когда я запускаю обе установки, иногда вторая ничего не печатает, иногда это происходит; Вероятно, вы что-то видите или нет.

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

Если вы положили Thread.sleep(1000), вы увидите, что оба печатают значение.

Answer 2

Фактически оба они печатают результат. Единственное различие - время. Это то, что я считаю объяснением наблюдаемого поведения. В первом случае flatMap напрямую подписался на результат just на main нить и println на новую тему. Нерест новой нити и println оба являются интенсивными операциями, требующими времени для выполнения, и до сих пор заканчивается нить main.

Во втором случае, flatMap выписывает just на новом потоке и println выписывает flatMap на main нити. Таким образом, поток main удерживается более занятым из-за зависимости после нереста нового потока, и мы видим результат до выхода main.

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

Observable.timer(3, TimeUnit.SECONDS).toBlocking().subscribe();