考慮以下代碼：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);您可以看到一個(gè)發(fā)布者比另一個(gè)發(fā)布者快 100 倍。不過，在運(yùn)行代碼時(shí)，似乎所有數(shù)據(jù)都已打印，但兩個(gè)發(fā)布者之間存在巨大差距，這會(huì)增加加班時(shí)間。有趣的是，當(dāng)更改數(shù)字時(shí)executer2會(huì)有1024線程，并且executer1只有1線程，然后我們?nèi)匀粫?huì)看到隨著時(shí)間的推移越來越大的差距。我的期望是，在相應(yīng)地調(diào)整線程池之后，發(fā)布者將得到平衡。我想在發(fā)布者之間取得平衡（相對(duì)于線程池大小和處理時(shí)間）如果我等的時(shí)間足夠長會(huì)發(fā)生什么？換句話說，是否會(huì)發(fā)生背壓？（默認(rèn)情況下，我猜這是一個(gè)運(yùn)行時(shí)異常，對(duì)吧？）我不想丟棄物品，也不想出現(xiàn)運(yùn)行時(shí)異常。相反，正如我提到的，我希望系統(tǒng)在其擁有的資源和處理時(shí)間方面取得平衡——上面的代碼是否承諾了這一點(diǎn)？
1 回答
暮色呼如
TA貢獻(xiàn)1853條經(jīng)驗(yàn) 獲得超9個(gè)贊
本例中的對(duì)象Flux不是ParallelFlux對(duì)象，因此它們只會(huì)使用一個(gè)線程。
如果您創(chuàng)建一個(gè)能夠處理數(shù)千個(gè)線程的調(diào)度程序，并將其傳遞給其中一個(gè)對(duì)象，這并不重要Flux——它們只會(huì)坐在那里不被使用，這正是本示例中發(fā)生的情況。沒有背壓，也不會(huì)導(dǎo)致異?！鼤?huì)盡可能快地使用一個(gè)線程。
如果要確保Flux充分利用可用的 1024 個(gè)線程，則需要調(diào)用.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)));
如果你對(duì)你的代碼這樣做，那么你會(huì)開始看到更接近你似乎期望的結(jié)果，盡管它的休眠時(shí)間是原來的 100 倍，但還是會(huì)過去2_：1_
...
2_17075
2_17076
1_863
1_864
2_17077
1_865
2_17078
2_17079
...
但是，請(qǐng)注意：
我想在發(fā)布者之間取得平衡（相對(duì)于線程池大小和處理時(shí)間）
你不能在這里選擇數(shù)字來平衡輸出，至少不能可靠地或以任何有意義的方式——線程調(diào)度將是完全任意的。如果你想這樣做，那么你可以使用subscribe 方法的這個(gè)變體，允許你顯式地調(diào)用request()訂閱消費(fèi)者。這樣一來，您就可以通過僅請(qǐng)求您準(zhǔn)備處理的盡可能多的元素來提供背壓。
添加回答
舉報(bào)
0/150
提交
取消