flux.take (x) veya flux.all (..) tüm emisyonları beklemez

Question

Iterable 8 öğesinden() oluşturulan bir flux var. Her bir akı emisyonu için, bir senkronize olmayan bir yöntemi çağırmak istiyorum. Ben çalışmak ve sonunda ben fluxflux<CompletableFuture<Boolean>> dönüştüğü map(CompletableFuture.supplyAsync(..), executor) ile yerleşmiş vermedi dispatchOn ve publishOn ile çeşitli yollar denedik.

Şimdi yalnızca son öğe tamamlandığında akışa devam etmek istiyorum. all(..) ve take(size of the Iterable) ile denedim ancak her iki durumda da tüm elemanlar tamamlanmadan önce akış devam ediyor. Bunun, yürütücümün yalnızca 4 iş parçacığı olduğu ve CompletableFuture -s akışına eklenmesinin biraz zaman alması nedeniyle olduğunu farz ediyorum.

Neden all(..) veya take(8) akı bitmiyor? Nasıl bekleyebilirim?

kodu: Her öğe için bir hesaplama ile paralel gitmek istiyorsanız

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

, sen flatMap + sadece + subscribeOn + haritayı kullanabilirsiniz. Sana RxJava bir örnek vereyim ben Reactor'ün zamanlayıcı türleri ile çok aşina değilim: Ben eklemek denemek istiyorum

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

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

Answer 2

senin sana oluşturamazsınız varsayabiliriz Çünkü

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

gözlemlenebilir yineleyici başına bir gözlemci, daha sonra hepsini birleştir ya da sıkıştır ve sadece abone ol.

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