如何查找Flux（父级）中的所有数据都由其内部非阻塞Flux或Mono（子级）处理？

Question

我有一个聚合器实用程序类，在那里我必须连接多个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();
            });
        }

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

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

在处理完Flux内部所有嵌套的Mono/Flux(非阻塞)请求后，我想调用onComplete。

对于嵌套阻塞磁通量/单晶通量，外部通量 doOn Complete 方法在完成内部通量/单晶后执行。

后记：-

在下面的例子中，我找不到放置emitter.complete（）的位置。因为doOnCompl（）方法是在完成所有内部Mono之前调用的。

请求正文：-

[{ "content":"Intro to React and operators", "author":"Josh Long", "name":"Spring WebFlux" },{ "content":"Intro to Flux", "author":"Josh Long", "name":"Spring WebFlux" },{ "content":"Intro to Mono", "author":"Josh Long", "name":"Spring WebFlux" }] 

我的Rest控制器：-

@PostMapping("/topics")
    public Flux<?> loadTopic(@RequestBody Flux<Alltopics> data)
    {
        return Flux.create(emitter ->{
            data
            .map(topic -> {
                topic.setTopicid(null ==topic.getTopicid() || topic.getTopicid().isEmpty()?UUID.randomUUID().toString():topic.getTopicid());
                return topic;
            })
            .doOnNext(topic -> {
                topicRepository.save(topic).doOnSuccess(persistedTopic ->{
                    emitter.next(persistedTopic);
                    //emitter.complete();
                }).subscribe();
            })
            .doOnComplete(() -> {
                //emitter.complete();
                System.out.println(" all the data are processed!!!");
            }).subscribe();
        });
    }

Answer 1

以下是编写反应式管道时应该遵循的一些规则：

1. doOnXYZ运算符永远不应该用于执行大量I/O、涉及延迟的操作或任何反应式操作。这些应该用于“副作用”操作，例如日志记录、指标等。
2. 您永远不应该从管道或返回反应类型的方法中订阅。这将此操作的处理与主管道分离，这意味着不能保证您会在正确的时间获得预期的结果，也不能保证您的应用程序会知道完整的/错误信号。
3. 您永远不应该在管道或返回反应式类型的方法中阻止。这将在运行时给您的应用程序带来严重问题。

现在，由于您的代码片段相当复杂，我将只给出使用另一个代码片段的大致方向。

@GetMapping("/info")
public Flux<FullDetails> getData(){
    return followersRepository.findAll()
        .flatMap(follower -> {
            Mono<List<Alltopics>> topics = topicRepository.findAllById(follower.getTopiclist()).collectList();
            return topics.map(topiclist -> new FullDetails(follower.getId(), topiclist));
        });
}