RxJava Scheduler ändert Thread nicht mit Schlaf

Question

Ich bin sehr seltsam konfrontiert RxJava Verhalten kann ich nicht verstehen.

Angenommen, ich möchte die Verarbeitung von Elementen parallelisieren. Ich verwende flatMap dafür:

public static void log(String msg) { 
    String threadName = Thread.currentThread().getName(); 
    System.out.println(String.format("%s - %s", threadName, msg)); 
} 

public static void sleep(int ms) { 
    try { 
     Thread.sleep(ms); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 
} 

public static void main(String[] args) throws InterruptedException { 

    Scheduler sA = Schedulers.from(Executors.newFixedThreadPool(1)); 
    Scheduler sB = Schedulers.from(Executors.newFixedThreadPool(5)); 

    Observable.create(s -> { 
     while (true) { 
      log("start"); 
      s.onNext(Math.random()); 
      sleep(10); 
     } 
    }).subscribeOn(sA) 
      .flatMap(r -> Observable.just(r).subscribeOn(sB)) 
      .doOnNext(r -> log("process")) 
      .subscribe((r) -> log("finish")); 
} 

der Ausgang ist ziemlich vorhersehbar:

pool-1-thread-1 - start 
pool-2-thread-1 - process 
pool-2-thread-1 - finish 
pool-1-thread-1 - start 
pool-2-thread-2 - process 
pool-2-thread-2 - finish 
pool-1-thread-1 - start 
pool-2-thread-3 - process 
pool-2-thread-3 - finish 

Na ok, aber wenn ich mit n Schlaf hinzufügen> 10 auf die Karte nach flatMap Parallelisierung Scheduler nicht mehr ändert Threads .

public static void main(String[] args) throws InterruptedException { 

    Scheduler sA = Schedulers.from(Executors.newFixedThreadPool(1)); 
    Scheduler sB = Schedulers.from(Executors.newFixedThreadPool(5)); 

    Observable.create(s -> { 
     while (true) { 
      log("start"); 
      s.onNext(Math.random()); 
      sleep(10); 
     } 
    }).subscribeOn(sA) 
      .flatMap(r -> Observable.just(r).subscribeOn(sB)) 
      .doOnNext(r -> sleep(15)) 
      .doOnNext(r -> log("process")) 
      .subscribe((r) -> log("finish")); 
} 

Was folgendes ergibt:

pool-1-thread-1 - start 
pool-1-thread-1 - start 
pool-2-thread-1 - process 
pool-2-thread-1 - finish 
pool-1-thread-1 - start 
pool-1-thread-1 - start 
pool-2-thread-1 - process 
pool-2-thread-1 - finish 
pool-1-thread-1 - start 
pool-2-thread-1 - process 

WARUM ??? Warum werden alle Elemente nach flatMap in demselben Thread (pool-2-thread-1) bearbeitet?

Answer 1

FlatMap serialisiert alle parallelen Aufgaben in einen einzelnen Thread und Sie spähen in diesen Thread. Versuchen Sie dies stattdessen

public static void main(String[] args) throws InterruptedException { 

Scheduler sA = Schedulers.from(Executors.newFixedThreadPool(1)); 
Scheduler sB = Schedulers.from(Executors.newFixedThreadPool(5)); 

Observable.create(s -> { 
    while (!s.isUnsubscribed()) { 
     log("start"); 
     s.onNext(Math.random()); 
     sleep(10); 
    } 
}).subscribeOn(sA) 
     .flatMap(r -> 
      Observable.just(r) 
      .subscribeOn(sB) 
      .doOnNext(r -> sleep(15)) 
      .doOnNext(r -> log("process")) 
     ) 
     .subscribe((r) -> log("finish")); 
}