使用spring数据mongodb reactive管理多个Flux的生命周期

Question

我有一个数据服务，我正在认真考虑切换到反应式模型。这是一个联合查询引擎，它可以通过调用一个或多个"；解析器"；实现，具体取决于查询类型。

如果我切换到spring-data-mongodb-reactive，那么这些实现中的每一个都必须为创建多个Flux实例

1. 对信息不同部分的查询
2. 为#1中的每个查询查询所有数据库

注意 ：我不想合并每个Flux，因为能够将上面#1的查询分开，可以使最终处理更容易。将每个"；部分"；对所有联邦数据库的查询都可以，但我必须保留每个"联邦"数据库的数据；部分"；分离我希望这是有道理的。

解释完整的工作流程不在本文的范围内，但我想知道如何创建任意数量的Flux实例，并订阅它们以启动它们，但要等到它们全部完成后，才能继续处理所有联邦源中完全检索到的数据。在Java中，我正在寻找类似于CompletableFuture.allOf()的东西。

如果我做这样的事情，我会接近正轨吗：

public class ReactiveDataService {
private static final Supplier<Example<String>> example1 = () -> Example.of("Example 1");
private static final Supplier<Example<String>> example2 = () -> Example.of("Example 2");
private static final Supplier<Example<String>> example3 = () -> Example.of("Example 3");
private static final Supplier<Example<String>> example4 = () -> Example.of("Example 4");
private static final Supplier<Example<String>> example5 = () -> Example.of("Example 5");
private final Collection<ReactiveMongoRepository<String, String>> repositories;
public ReactiveDataService(Collection<ReactiveMongoRepository<String, String>> repositories) {
this.repositories = repositories;
}
private void processFluxes(final Flux<String> flux1, final Flux<String> flux2, final Flux<String> flux3,
final Flux<String> flux4, final Flux<String> flux5) {
// Call service to process flux stuff
}
/**
* For all repositories, combine fluxes that run the same query.
* Subscribe to each flux immediately to get the query started.
* Add all fluxes to a container flux that processes the results
* upon completion.
* After everything is set up, block until completion.
*/
public void doQuery() {
final Flux<String> flux1 = Flux.fromIterable(repositories)
.flatMap(repo -> repo.findAll(example1.get()));
flux1.subscribe();
final Flux<String> flux2 = Flux.fromIterable(repositories)
.flatMap(repo -> repo.findAll(example2.get()));
flux2.subscribe();
final Flux<String> flux3 = Flux.fromIterable(repositories)
.flatMap(repo -> repo.findAll(example3.get()));
flux3.subscribe();
final Flux<String> flux4 = Flux.fromIterable(repositories)
.flatMap(repo -> repo.findAll(example4.get()));
flux4.subscribe();
final Flux<String> flux5 = Flux.fromIterable(repositories)
.flatMap(repo -> repo.findAll(example5.get()));
flux5.subscribe();
final Flux<Flux<String>> fluxes = Flux.just(flux1, flux2, flux3, flux4, flux5)
.doOnComplete(() -> processFluxes(flux1, flux2, flux3, flux4, flux5));
fluxes.blockLast();
}
}

Answer 1

以下是如何使用Mono.zip的示例：

public static void main(String[] args) {
Flux<String> flux0 = Flux.empty();
Flux<String> flux1 = Flux.just("1.1", "1.2", "1.3");
Flux<String> flux2 = Flux.just("2.1", "2.2", "2.3");
Flux<String> flux3 = Flux.just("3.1", "3.2", "3.3");

Mono.zip(lists -> process(lists), flux0.collectList(), flux1.collectList(), flux2.collectList(), flux3.collectList()).block();
}

private static String process(Object[] lists) {
System.out.println("List 0 is " + lists[0]);
System.out.println("List 1 is " + lists[1]);
System.out.println("List 2 is " + lists[2]);
System.out.println("List 3 is " + lists[3]);
return "output";
}

输出：

List 0 is []
List 1 is [1.1, 1.2, 1.3]
List 2 is [2.1, 2.2, 2.3]
List 3 is [3.1, 3.2, 3.3]

所以你可以根据自己的情况进行调整。

请注意，Mono.zip不能返回null，这就是为什么我将"输出"；因此，但如果您不需要任何输出，您可以放置任何不为空的内容。

其思想是首先使用collectList将每个Flux<String>转换为Mono<List<String>>，这样处理起来会更简单。Mono.zip允许您等待所有操作完成，并将输出处理为Object[]。您可以将每个对象转换为一个List<String>进行处理。