如何将 Mono 和 Flux 结合起来

Question

我正在创建一个产品 Flux 和一个需要应用于产品的 Mono Category 对象，因为它们都被分组到同一类别中。我正在尝试像这样将类别添加到每个产品。

     Flux<Product> products = productRepository.findAllById(productCodes);
     Mono<Category> category = categoryRepository.findById(productCodes.get(0));
     
     The category for all these products should be set as:

     //If these were a list of products i.e List<Product> productList and Category was 
     //not a Mono I would set them like this:
 
     productList.forEach( product -> product.setCategory(category));

由于各自的存储库分别返回 Mono 和 Flux，我不确定如何使用反应式框架来做到这一点，因为我对它非常陌生。有人可以告诉我如何使用反应式来实现这一点吗？

Answer 1

实际上，有很多方法可以实现这一点。

这是一个示例，您将带有

flatMap()

的函数应用于产品通量的每个元素。此功能使用您的类别

Mono

来将类别设置为产品。

productsFlux.flatMap(product -> categoryMono.map(category -> {
    product.setCategory(category);
    return product;
}));

或者反之亦然，在 Mono 类别上使用

flatMapMany()

：

categoryMono.flatMapMany(category -> productsFlux.map(product -> {
    product.setCategory(category);
    return product;
}));

或者您可以将您的类别

Mono

变成带缓存的热永无止境的通量，并对您的产品通量和类别通量执行

Flux.zip()

，如下所示：

Flux<String> categoryHotFlux = categoryMono.cache().repeat();

Flux.zip(productsFlux, categoryHotFlux)
        .map(tuple -> {
            Product product = tuple.getT1();
            Category category = tuple.getT2();
            product.setCategory(category);
            return product;
        });