如何找到 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();
});
}

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

我已经尝试使用doOnComplete和doOnLast进行外部Flux，它不会等待所有内部非阻塞调用完成。

我想在处理 Flux 内所有嵌套的单声道/通量(非阻塞(请求后调用 onComplete。

对于嵌套阻塞通量/单声道，外部助焊剂doOnComplete方法在完成内部助焊剂/单声道后执行。

---

后记(PS(：-

在下面的示例中，我找不到放置 emitter.complete(( 的位置。 因为doOnComplete((方法在所有内部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" }] 

我的休息控制器：-

@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. 切勿从返回反应式类型的管道或方法中subscribe。这会将此操作的处理与主管道分离，这意味着无法保证您会在正确的时间获得预期的结果，也不能保证您的应用程序会知道完整/错误信号。
3. 切勿从返回反应式类型的管道或方法中block。这将在运行时给应用程序带来严重问题。

现在，由于您的代码片段非常复杂，我将只为您提供另一个代码片段的一般方向。

@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));
});
}