RxJava - From Scheduler

RxJava - From Scheduler

Executor Scheduler — This is more of a custom IO Scheduler. In this scheduler, we can create a custom pool of threads by specifying the size of pool. This can be used in scenarios where number of observables could be huge for IO Schedulers. Below is an example of Executor Scheduler

Example:

public class RxSchedulers {
    
    public static void main(String[] args) throws InterruptedException
    {
        
         Observable.just("Rx", "Java", "Scheduler")
         .flatMap(v -> getLengthWithDelay(v)
         .doOnNext(s -> System.out.println("Processing Thread " 
            + Thread.currentThread().getName()))
         .subscribeOn(Schedulers.from(Executors.newFixedThreadPool(3))))
         .subscribe(length -> System.out.println("Receiver Thread " 
            + Thread.currentThread().getName() 
            + ", Item length " + length));

         Thread.sleep(10000);
   }
   protected static Observable<Integer> getLengthWithDelay(String v) {
      Random random = new Random();
      try {
         Thread.sleep(random.nextInt(3) * 1000);
         return Observable.just(v.length());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return null;
   }

}

 

Output

Processing Thread pool-1-thread-1
Receiver Thread pool-1-thread-1, Item length 2
Processing Thread pool-3-thread-1
Receiver Thread pool-3-thread-1, Item length 4
Processing Thread pool-4-thread-1
Receiver Thread pool-4-thread-1, Item length 9