flux.take（x）またはflux.all（..）がすべての排出を待つことはありません

Question

私はIterableの8要素（Flux.fromIterable(..)）から構築されたfluxを持っています。 各フラックスの排出量ごとに、メソッドを非同期に呼び出す必要があります。 dispatchOnとpublishOnでうまくいきませんでしたが、最終的にmap(CompletableFuture.supplyAsync(..), executor)で解決してfluxをflux<CompletableFuture<Boolean>>に変換しました。

最後の項目が完了したときにのみフローを続行します。 all(..)と、take(size of the Iterable)を試しましたが、どちらの場合も、すべての要素が完了する前にフローが継続します。 これは、私のエグゼキュータがスレッドを4つしか持たず、CompletableFuture -sをフラックスに追加するまでに時間がかかるからです。

all(..)またはtake(8)なぜフラックスが完了するのを待っていますか？ どうすれば待つことができますか？

コード：

Mono 
    .fromFuture(dbUtil.getEntity(id)) 
    .doOnError(t -> { 
     ... 
     return;}) 
    .doOnSuccess(s -> log.info("Got it: " + s)) 
    .flatMap(s -> 
     Flux.fromIterable(s.getItemsMap().entrySet()) 
      .map(e -> CompletableFuture.supplyAsync(()->process(e, s), EXECUTOR)) 
      .take(s.getItemsMap().entrySet().size()) 
    ) 
    .all(...) 
    .consume(b -> done(b)); 

Answer 1

あなたは、各項目のための計算と並行に行きたい場合は、flatMap +ちょうど+ subscribeOn +マップを使用することができます。私はRxJavaであなたに例をあげるので、私は、原子炉のスケジューラタイプとあまり慣れていないよ：

ExecutorService exec = ... 
Scheduler scheduler = Schedulers.from(exec); 

Observable.from(...) 
.flatMap(e -> Observable.just(e).subscribeOn(scheduler).map(v -> process(v))) 
.subscribe(...); 

Answer 2

私は私はあなたが作成することができないと仮定することができますので、あなたに観察

.finallyDo("here arrive when onComplete or onError") 

に追加しようとするだろうイテレータごとに観測可能なものを作成し、それらをマージまたはジップしてサブスクライブするだけです。

Observable.merge(obsevableIter1(), observableIter2()) 
       .subscribe(result -> showResult(result.toString()));