考慮以下代碼：AtomicInteger counter1 = new AtomicInteger();AtomicInteger counter2 = new AtomicInteger();Flux<Object> source = Flux.generate(emitter -> { emitter.next("item");});Executor executor1 = Executors.newFixedThreadPool(32);Executor executor2 = Executors.newFixedThreadPool(32);Flux<String> flux1 = Flux.merge(source).concatMap(item -> Mono.fromCallable(() -> { Thread.sleep(1); return "1_" + counter1.incrementAndGet();}).subscribeOn(Schedulers.fromExecutor(executor1)));Flux<String> flux2 = Flux.merge(source).concatMap(item -> Mono.fromCallable(() -> { Thread.sleep(100); return "2_" + counter2.incrementAndGet();}).subscribeOn(Schedulers.fromExecutor(executor2)));Flux.merge(flux1, flux2).subscribe(System.out::println);您可以看到一個發布者比另一個發布者快 100 倍。不過，在運行代碼時，似乎所有數據都已打印，但兩個發布者之間存在巨大差距，這會增加加班時間。有趣的是，當更改數字時executer2會有1024線程，并且executer1只有1線程，然后我們仍然會看到隨著時間的推移越來越大的差距。我的期望是，在相應地調整線程池之后，發布者將得到平衡。我想在發布者之間取得平衡（相對于線程池大小和處理時間）如果我等的時間足夠長會發生什么？換句話說，是否會發生背壓？（默認情況下，我猜這是一個運行時異常，對吧？）我不想丟棄物品，也不想出現運行時異常。相反，正如我提到的，我希望系統在其擁有的資源和處理時間方面取得平衡——上面的代碼是否承諾了這一點？
1 回答
暮色呼如
TA貢獻1853條經驗 獲得超9個贊
本例中的對象Flux不是ParallelFlux對象，因此它們只會使用一個線程。
如果您創建一個能夠處理數千個線程的調度程序，并將其傳遞給其中一個對象，這并不重要Flux——它們只會坐在那里不被使用，這正是本示例中發生的情況。沒有背壓，也不會導致異?！鼤M可能快地使用一個線程。
如果要確保Flux充分利用可用的 1024 個線程，則需要調用.parallel(1024)：
ParallelFlux<String> flux1 = Flux.merge(source).parallel(1).concatMap(item -> Mono.fromCallable(() -> {
Thread.sleep(1);
return "1_" + counter1.incrementAndGet();
}).subscribeOn(Schedulers.fromExecutor(executor1)));
ParallelFlux<String> flux2 = Flux.merge(source).parallel(1024).concatMap(item -> Mono.fromCallable(() -> {
Thread.sleep(100);
return "2_" + counter2.incrementAndGet();
}).subscribeOn(Schedulers.fromExecutor(executor1)));
如果你對你的代碼這樣做，那么你會開始看到更接近你似乎期望的結果，盡管它的休眠時間是原來的 100 倍，但還是會過去2_：1_
...
2_17075
2_17076
1_863
1_864
2_17077
1_865
2_17078
2_17079
...
但是，請注意：
我想在發布者之間取得平衡（相對于線程池大小和處理時間）
你不能在這里選擇數字來平衡輸出，至少不能可靠地或以任何有意義的方式——線程調度將是完全任意的。如果你想這樣做，那么你可以使用subscribe 方法的這個變體，允許你顯式地調用request()訂閱消費者。這樣一來，您就可以通過僅請求您準備處理的盡可能多的元素來提供背壓。
添加回答
舉報
0/150
提交
取消