RxJava: Использование SubscribeOn делает выход программы без завершения

Question

Это мой пример код

Observable.range(1,5) 
      .subscribeOn(Schedulers.computation()) 
      .map(Observables05::doSomething) 
      .subscribe(System.out::println, Throwable::printStackTrace,() -> System.out.println("done")); 

Моего йоЗотеЬЫпд Метод,

public static int doSomething(int i) { 
    try { 
     System.out.println("Processing " + i + 
       " on Thread -- " + Thread.currentThread().getName()); 
     Thread.sleep(500); 
     return i; 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
     throw new RuntimeException(e); 
    } 
} 

Используя только образец кода в моем основном потоке просто выходит из программы , Однако, если после этого используется Thread.sleep(3000), тогда программа работает правильно, прежде чем выйти, когда закончится время ожидания.

Является ли это ожидаемым поведением и почему? Как я могу сделать этот код без использования Thread.sleep?

Answer 1

subscribeOn графики subscribe звонок в фоновый поток (вы выбрали computation планировщик). После этого планирования ваш основной поток будет свободен, т. Е. Прекратит работу вашей программы.

Как-то вам нужно дождаться завершения всех желаемых задач перед выходом. Thread.sleep(3000) выполняет работу для простых тестовых случаев.

Реальные программы обычно не прекращаются так быстро. Однако есть случаи, когда вам нужно дождаться завершения какой-либо фоновой задачи. Существуют различные механизмы синхронизации потоков (например, CountDownLatch), которые вы можете использовать для этого.

Answer 2

Если вы хотите протестировать код, который вы написали, используйте TestSubscriber.

код будет выглядеть следующим образом:

TestSubscriber<Integer> testSubscriber = new TestSubscriber<>(); 

Observable.range(1,5) 
     .subscribeOn(Schedulers.computation()) 
     .map(Observables05::doSomething) 
     .subscribe(testSubscriber); 

testSubscriber.awaitTerminalEvent(10, TimeUnit.SECONDS); 

Имейте в виду, что TestSubscriber позволяет тестировать намного больше, чем только терминала события.

Answer 3

Это ожидаемое поведение. Простое решение для вас, как вы в выполнении main метода заключается в использовании toBlocking(), прежде чем подписаться - как показано ниже -

Observable.range(1,5) 
    .subscribeOn(Schedulers.computation()) 
    .map(DummyJunk::doSomething) 
    .toBlocking() 
    .subscribe(System.out::println, Throwable::printStackTrace,() -> System.out.println("done"));