RxJava - 爲什麼執行者只使用一個線程

Question

我已經創建了一個固定的線程池來處理每300毫秒發送一個事件，並假定該進程需要1000毫秒。假設多線程將工作，但只有一個線程被重用。

如果我設置的sleepTime小於300毫秒，處理線程改變，但是沒用。

問題：我能做些什麼來使它併發？爲什麼程序重用線程？

預先感謝您

public static void main(String[] args) throws InterruptedException { 
    long sleepTime = 1000; 
    ExecutorService e = Executors.newFixedThreadPool(3); 

    Observable.interval(300, TimeUnit.MILLISECONDS) 
    .subscribeOn(Schedulers.computation()) 
    .flatMap(new Func1<Long, Observable<Long>>() { 
     @Override 
     public Observable<Long> call(Long pT) { 
      return Observable.just(pT).subscribeOn(Schedulers.from(e)); 
     } 
    }) 
    .doOnNext(new Action1<Long>() { 

     @Override 
     public void call(Long pT) { 
      try { 
       Thread.sleep(sleepTime); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 

     } 
    }) 
    .subscribe(new Action1<Long>() { 

     @Override 
     public void call(Long pT) { 
      System.out.println("i am " + pT + "in thread:" + Thread.currentThread().getName()); 

     } 
    }); 


    Thread.sleep(50000); 
    e.shutdownNow(); 

} 

日誌

i am 0in thread:pool-1-thread-1 
i am 1in thread:pool-1-thread-1 
i am 2in thread:pool-1-thread-1 
i am 3in thread:pool-1-thread-1 
i am 4in thread:pool-1-thread-1 
i am 5in thread:pool-1-thread-1 
i am 6in thread:pool-1-thread-1 
i am 7in thread:pool-1-thread-1 
i am 8in thread:pool-1-thread-1 
i am 9in thread:pool-1-thread-1 
i am 10in thread:pool-1-thread-1 
i am 11in thread:pool-1-thread-1 

Answer 1

據我瞭解在你的代碼，生產者在比用戶更快的速度生產。但是Observable<Long> interval(long interval, TimeUnit unit)實際上不支持Backpressure。文檔指出

該操作員不支持背壓，因爲它使用時間。如果下游的 需要較慢，則應該減慢計時器或使用類似於{@link #onBackpressureDrop}的東西 。

如果你的處理真的比生產速度較慢，你可以在你的用戶代碼做的就是這樣的事情

.subscribe(new Action1<Long>() { 

    @Override 
    public void call(Long pT) { 
     e.submit(new Runnable() { 
      System.out.println("i am " + pT + "in thread:" + Thread.currentThread().getName()); 

     } 
    } 
}); 

Answer 2

相反

.subscribeOn(Schedulers.computation()) 

嘗試

.observeOn(Schedulers.computation()) 

這個我以前玩過機智的例子h與Rx的併發工作相當不錯，例如

public class ObservableZip { 

private Scheduler scheduler; 
private Scheduler scheduler1; 
private Scheduler scheduler2; 

@Test 
public void testAsyncZip() { 
    scheduler = Schedulers.newThread(); 
    scheduler1 = Schedulers.newThread(); 
    scheduler2 = Schedulers.newThread(); 
    long start = System.currentTimeMillis(); 
    Observable.zip(obAsyncString(), obAsyncString1(), obAsyncString2(), (s, s2, s3) -> s.concat(s2) 
                         .concat(s3)) 
       .subscribe(result -> showResult("Async:", start, result)); 
} 




public void showResult(String transactionType, long start, String result) { 
    System.out.println(result + " " + 
           transactionType + String.valueOf(System.currentTimeMillis() - start)); 
} 


public Observable<String> obAsyncString() { 
    return Observable.just("") 
        .observeOn(scheduler) 
        .doOnNext(val -> { 
         System.out.println("Thread " + Thread.currentThread() 
                   .getName()); 
        }) 
        .map(val -> "Hello"); 
} 

public Observable<String> obAsyncString1() { 
    return Observable.just("") 
        .observeOn(scheduler1) 
        .doOnNext(val -> { 
         System.out.println("Thread " + Thread.currentThread() 
                   .getName()); 
        }) 
        .map(val -> " World"); 
} 

public Observable<String> obAsyncString2() { 
    return Observable.just("") 
        .observeOn(scheduler2) 
        .doOnNext(val -> { 
         System.out.println("Thread " + Thread.currentThread() 
                   .getName()); 
        }) 
        .map(val -> "!"); 
    } 

} 

Answer 3

我在GitHub上找到了答案！

內部observable確實發射在多線程上，但接下來做的不是。如果我想讓它平行，我應該在內部可觀察的情況下做。