Project Reactor：在Mono中将Flux聚合为List属性

在project reactor等响应式编程框架中，数据以异步流的形式（flux表示0到n个元素，mono表示0到1个元素）进行处理。然而，在实际开发中，我们经常需要将这些异步流中的数据聚合起来，并将其赋给传统java对象（pojo）的属性，特别是当该属性是一个集合类型（如list）时。

一个常见的场景是：我们从服务层获取到一个Flux，代表一系列异步到达的Item对象。同时，我们有一个Mono，其中Person对象包含一个List类型的属性。此时，我们面临的问题是如何将这个Flux中的所有Item收集起来，并将其赋值给Mono内部Person对象的items列表属性。

直接尝试将Flux赋值给List会导致编译错误，因为它们的类型不匹配。Flux是一个数据发布者，而List是一个具体的数据结构。为了解决这个问题，我们需要一种机制来“等待”Flux完成所有元素的发布，然后将这些元素收集到一个List中，最终将这个List安全地嵌入到Mono包装的Person对象中。

核心解决方案：collectList()与map操作符

Project Reactor提供了强大的操作符来处理这类场景。解决上述问题的关键在于两个操作符的组合使用：

1. collectList(): 这是Flux上的一个操作符，它的作用是将Flux发出的所有元素收集到一个List中，并将其包装成一个Mono>返回。这意味着collectList()会等待Flux完成（即所有元素都被发出），然后将收集到的列表作为单个元素发布到下游的Mono中。
2. map(): 这是Mono上的一个操作符，它允许我们对Mono内部的值进行同步转换。当Mono>中的List可用时，map()操作符可以接收这个List，并将其转换为我们期望的Mono。

通过这两个操作符的组合，我们可以构建一个清晰的响应式处理链，实现将Flux聚合为List并嵌入到Mono中的目标。

实战演练：构建Mono并填充List

为了演示这个过程，我们首先定义所需的POJO类和模拟服务：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

// Item 类定义
class Item {
    private String name;

    public Item(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "Item{" + "name='" + name + '\'' + '}';
    }
}

// Person 类定义，包含一个 List 属性
class Person {
    private List items;

    public Person() {
        // 可以在构造函数中初始化列表，或者在设置时处理
    }

    public Person(List items) {
        this.items = items;
    }

    public List getItems() {
        return items;
    }

    public void setItems(List items) {
        this.items = items;
    }

    @Override
    public String toString() {
        return "Person{" + "items=" + items + '}';
    }
}

// 模拟服务层接口，返回 Flux
interface ItemService {
    Flux getItems();
}

// ItemService 的具体实现
class MyItemService implements ItemService {
    @Override
    public Flux getItems() {
        // 模拟异步获取 Item 列表，每个 Item 之间有延迟
        return Flux.just(new Item("Laptop"), new Item("Mouse"), new Item("Keyboard"))
                   .delayElements(Duration.ofMillis(100)); // 模拟异步延迟
    }
}

public class FluxToListInMonoExample {

    private final ItemService itemService = new MyItemService(); // 注入服务

    /**
     * 创建一个 Mono，其中 Person 对象的 items 属性通过聚合 Flux 得到。
     *
     * @return 包含聚合后 Item 列表的 Mono
     */
    public Mono createPersonWithCollectedItems() {
        // 1. 从服务层获取一个 Flux 数据流
        Flux itemFlux = itemService.getItems();

        // 2. 使用 collectList() 操作符将 Flux 聚合成 Mono>
        //    这个 Mono 会在 itemFlux 发出所有 Item 后，发布一个包含所有 Item 的 List。
        Mono> collectedItemsMono = itemFlux.collectList();

        // 3. 使用 map() 操作符将 Mono> 转换为 Mono
        //    当 List 可用时，创建一个 Person 对象并设置其 items 属性。
        Mono personMono = collectedItemsMono.map(itemList -> {
            Person person = new Person(); // 创建一个新的 Person 实例
            person.setItems(itemList);    // 将收集到的 List 设置给 Person 对象
            return person;                // 返回包含 List 的 Person 对象
        });

        return personMono;
    }

    public static void main(String[] args) {
        FluxToListInMonoExample example = new FluxToListInMonoExample();

        System.out.println("开始聚合 Item 并创建 Person 对象...");
        example.createPersonWithCollectedItems()
                .doOnNext(person -> {
                    System.out.println("成功创建 Person 对象: " + person);
                    if (person.getItems() != null && !person.getItems().isEmpty()) {
                        System.out.println("包含的 Item 数量: " + person.getItems().size());
                        person.getItems().forEach(item -> System.out.println(" - " + item.getName()));
                    } else {
                        System.out.println("Person 对象不包含任何 Item 或列表为空。");
                    }
                })
                .doOnError(error -> System.err.println("处理过程中发生错误: " + error.getMessage()))
                .block(); // 阻塞等待结果，仅用于示例演示，生产代码中应避免使用 block()
        System.out.println("操作完成。");
    }
}

代码详解

1. Flux itemFlux = itemService.getItems();:

   

   DALL·E 2

   OpenAI基于GPT-3模型开发的AI绘图生成工具，可以根据自然语言的描述创建逼真的图像和艺术。

   下载
   • 这一步模拟从外部服务获取一个Item数据流。itemService.getItems()返回一个Flux，表示Item对象会随着时间异步地发出。

2. Mono> collectedItemsMono = itemFlux.collectList();:

   • 这是核心步骤。collectList()操作符订阅itemFlux。它会等待itemFlux发出所有Item，并将它们逐一添加到内部的一个List中。
   • 一旦itemFlux完成（即不再发出新的Item），collectList()就会将这个完整的List作为单个元素发布到下游的Mono>中。此时，我们拥有了一个包含所有Item的列表，并且这个列表被封装在一个Mono中。

3. Mono personMono = collectedItemsMono.map(itemList -> { ... });:

   • map()操作符作用于collectedItemsMono。当collectedItemsMono发布其内部的List时，map操作符的lambda表达式会被执行。
   • 在lambda表达式内部，我们接收到完整的itemList。此时，我们可以安全地创建一个新的Person对象，并将itemList赋值给它的items属性。
   • 最后，map操作符将这个新创建的Person对象包装成Mono并发布到下游。

4. main方法中的订阅和阻塞:

   • doOnNext()用于在Mono成功发布Person对象时执行一些副作用操作，例如打印结果。
   • doOnError()用于处理可能发生的错误。
   • block()是一个阻塞操作，它会暂停当前线程，直到Mono完成并发出其结果。在生产环境中，应尽量避免使用block()，因为它违背了响应式编程的非阻塞原则。 block()主要用于测试、演示或在需要将响应式流与传统阻塞代码桥接的特定场景。在实际应用中，通常会订阅Mono并返回它，让调用者处理订阅和后续操作。

注意事项与最佳实践

• 理解响应式流的语义: Flux和Mono代表的是“可能在未来某个时间点发生”的数据流，而不是立即可用的数据。因此，不能像操作普通Java对象一样直接访问其内部数据，必须通过操作符来处理。
• 选择合适的聚合操作符: collectList()适用于需要收集所有元素后再进行下一步操作的场景。如果只需要对每个元素进行操作，或者只需要收集特定数量的元素，可以考虑buffer()、window()或其他collect系列操作符。
• 不可变性: 在响应式编程中，推荐使用不可变对象。在map操作中，我们创建了一个新的Person实例并设置其列表，而不是修改一个已存在的Person实例。这有助于避免并发问题和提高代码可预测性。
• 错误处理: 在实际应用中，务必为响应式流添加适当的错误处理机制，例如onErrorResume、onErrorReturn等，以优雅地处理可能发生的异常。
• 避免阻塞: 如前所述，block()应该谨慎使用。在大多数WebFlux应用中，您会返回Mono或Flux，让框架来管理订阅和线程。

总结

通过Flux的collectList()操作符将异步元素聚合为Mono>，再结合Mono的map()操作符进行类型转换，我们可以优雅且高效地将响应式数据流中的集合数据集成到普通的POJO对象中。这种模式是Project Reactor中处理异步数据聚合和转换的常见且推荐的方式，它确保了代码的响应性和类型安全性。掌握这种模式对于构建健壮的响应式应用程序至关重要。