如何在一个有嵌套单声道(非阻塞)的流中找到所有被处理的数据？

Question

假设我有下面的通量和单声道嵌套。我有两个不同的Apache Cassandra表的信息。我想合并细节，并作为通量发送回。

请参阅下面更新的伪代码。

@Autowired FollowersRepository followersRepository;
@Autowired TopicRepository topicRepository;
    @GetMapping("/info")
    public Flux<FullDetails> getData(){
        return Flux.create(emitter ->{
            followersRepository.findAll() 
            .doOnNext(data -> {
                List<String> all = data.getTopiclist(); //will get list of topic id
                List<Alltopics> processedList = new ArrayList<Alltopics>();
                all.forEach(action -> {
                    topicRepository.findById(action) //will get full detail about topic
                    .doOnSuccess(topic ->{
                        processedList.add(topic);
                        if (processedList.size() >= all.size()) {
                            FullDetails fulldetails = new FullDetails(action,processedList);
                            emitter.next(fulldetails);
                            //emitter.complete();
                        }
                    })
                    .subscribe();
                });
            })
            .doOnComplete(() ->{
                System.out.println("All the data are processed !!!");
                //emitter.complete(); // executing if all the data are pushed from database not waiting for doOnNext method to complete.
            })
            .subscribe();
        });
    }

有关更多详细信息，请参阅此处的代码CodeLink。

我已经尝试了 doOnOn 完成和 doOnFin 对于外部 Flux，它不会等待所有内部非阻塞调用完成。

我想在处理完 Flux 内部的所有嵌套 Mono（非阻塞）请求后调用“完成”。

Answer 1

发射器内的Flux实际上没有做任何事情，因为没有订户。发射器通常会对引发的事件做出反应，例如收到消息等。您可以在下面添加subscribe（）以使其工作。阅读热与冷订阅者。http://projectreactor.io/docs/core/snapshot/reference/#reactor.hotCold

return Flux.create(emitter -> {
            Flux.just(1,2,3,4,5) //list of ids from database

                    .doOnNext(uuid ->{
                        this.getData(uuid).doOnSuccess((result) -> {
                            System.out.println("query data from database "+uuid);
                            emitter.next("Data from database.");
                        });
                    })
                    .doOnComplete(()->{
                        System.out.println("Not waiting for all the Nested Mono to complete. ");
                    })
            .subscribe();
        });

如果你调用的是数据库，你可能不需要担心通过发射器引发事件

public Flux<String> getAllData2(){
        return Flux.just(1, 2, 3, 4, 5)
                .flatMap(uuid1 -> getData(uuid1).doOnSuccess(result -> System.out.println("query data from database " + result)))
                .doOnComplete(() -> System.out.println("Not waiting for all the Nested Mono to complete. "));
    }