Проблема с подпиской на Observable в RxJava

Проблема с вашим кодом заключается в том, что вы подписываетесь на другой поток в качестве текущего потока (основного). Таким образом, ваш основной поток завершается до того, как будет вызвана подписка на отдельные элементы.

Вот пример кода с трассировкой стека:

System.out.println(Thread.currentThread().getName() + ": start...");

Observable<String> stream = Observable.fromIterable(List.of("a", "b", "c"));
stream.subscribeOn(Schedulers.io())
      .subscribe(next -> System.out.println(Thread.currentThread().getName() + ": " + next), 
                 error -> error.printStackTrace());

Thread.sleep(1000L);

System.out.println(Thread.currentThread().getName() + ": stop...");

Трассировки стека:

main: start...
RxCachedThreadScheduler-1: a
RxCachedThreadScheduler-1: b
RxCachedThreadScheduler-1: c
main: stop...

Показано, что наблюдаемая создается в потоке main. Выполнение из подписки выполняется в другом потоке RxCachedThreadScheduler-1.

Пример с выполнением в том же потоке:

System.out.println(Thread.currentThread().getName() + ": start...");

Observable<String> stream = Observable.fromIterable(List.of("a", "b", "c"));
stream.subscribe(next -> System.out.println(Thread.currentThread().getName() + ": " + next), 
                 error -> error.printStackTrace());

System.out.println(Thread.currentThread().getName() + ": stop...");

Трассировки стека:

main: start...
main: a
main: b
main: c
main: stop...

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

На вопрос, это поток демона? Ответ да. Я прикрепил тестовый код:

String message = String.format("%s-isDaemon?%s: start...", Thread.currentThread().getName(), Thread.currentThread().isDaemon());
System.out.println(message);

Observable<String> stream = Observable.fromIterable(List.of("a", "b", "c"));
stream.subscribeOn(Schedulers.io())
      .subscribe(next -> {
                 final String innerMessage = String.format("%s-isDaemon?%s: %s", Thread.currentThread().getName(), Thread.currentThread().isDaemon(), next);
                 System.out.println(innerMessage);
                 },
                 Throwable::printStackTrace);

Thread.sleep(1000L);

message = String.format("%s-isDaemon?%s: stop...", Thread.currentThread().getName(), Thread.currentThread().isDaemon());
System.out.println(message);

Вывод стека:

main-isDaemon?false: start...
RxCachedThreadScheduler-1-isDaemon?true: a
RxCachedThreadScheduler-1-isDaemon?true: b
RxCachedThreadScheduler-1-isDaemon?true: c
main-isDaemon?false: stop...