将 Flux 转换为 Mono 中的 List：Reactor 教程

本文旨在解决在使用 Project Reactor 时，如何将一个 Flux 流转换为 List 并将其设置到 Mono 对象内部的问题。我们将详细介绍如何使用 collectList() 操作符将 Flux 收集为 List，然后将其映射到 Mono 对象中，最终创建一个包含完整数据的新 Person 对象。

在使用 Project Reactor 构建响应式应用时，经常会遇到需要将一个 Flux 流中的数据收集到一个 List 中，然后将这个 List 应用于某个 Mono 对象的情况。例如，你可能有一个 Person 类，其中包含一个 items 属性，该属性是一个 List。你希望从一个 Flux 中获取所有 Item 对象，并将它们添加到 Person 对象的 items 列表中。

以下是一个解决此问题的步骤和示例代码：

1. 理解 Reactor 操作符

首先，你需要理解几个关键的 Reactor 操作符：

• Flux: 代表一个包含 0 到 N 个元素的异步序列。
• Mono: 代表一个包含 0 或 1 个元素的异步序列。
• collectList(): Flux 的一个操作符，用于将 Flux 中的所有元素收集到一个 List 中，并返回一个 Mono>。
• map(): 用于将 Mono 或 Flux 中的每个元素转换为另一个元素。

2. 示例代码

假设我们有以下 Person 和 Item 类：

ONLYOFFICE

用ONLYOFFICE管理你的网络私人办公室

下载

import java.util.List;

public class Person {

    private List items;

    public List getItems() {
        return items;
    }

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

public 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;
    }
}

以及一个服务层方法，返回一个 Flux：

import reactor.core.publisher.Flux;

public class ItemService {

    public Flux getItems() {
        // 模拟从数据源获取 Item 列表
        return Flux.just(new Item("Item 1"), new Item("Item 2"), new Item("Item 3"));
    }
}

现在，我们可以编写代码将 Flux 转换为 List 并将其设置到 Mono 中：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;

public class PersonCreator {

    private final ItemService itemService;

    public PersonCreator(ItemService itemService) {
        this.itemService = itemService;
    }

    public Mono createPersonWithItems() {
        Flux itemsFlux = itemService.getItems(); // 获取 Item 流
        Mono> itemsListMono = itemsFlux.collectList(); // 将 Item 流收集为 List

        return itemsListMono.map(itemList -> { // 将 List 映射到 Person 对象
            Person person = new Person();
            person.setItems(itemList);
            return person;
        });
    }

    public static void main(String[] args) {
        ItemService itemService = new ItemService();
        PersonCreator personCreator = new PersonCreator(itemService);

        Mono personMono = personCreator.createPersonWithItems();

        personMono.subscribe(person -> {
            System.out.println("Person with items: " + person.getItems().size());
            person.getItems().forEach(item -> System.out.println("Item name: " + item.getName()));
        });
    }
}

代码解释：

1. itemService.getItems(): 从服务层获取 Flux。
2. itemsFlux.collectList(): 将 Flux 收集到一个 List 中，并返回一个 Mono>。 collectList() 会等待 Flux 完成，然后发出包含所有元素的 List。
3. itemsListMono.map(...): 使用 map 操作符将 Mono> 转换为 Mono。在 map 操作符中，我们创建一个新的 Person 对象，并将收集到的 List 设置到 Person 对象的 items 属性中。

3. 注意事项

• 错误处理： 在实际应用中，你需要考虑错误处理。例如，如果 itemService.getItems() 抛出异常，你需要使用 onErrorResume 或 onErrorReturn 等操作符来处理错误。
• 线程模型： Reactor 使用非阻塞的线程模型。确保你的代码不会阻塞 Reactor 的线程。
• 性能： 对于非常大的 Flux，collectList() 可能会消耗大量内存。在这种情况下，考虑使用其他操作符，例如 window 和 flatMap，来分批处理数据。
• 同步阻塞: 避免在Reactor链中使用阻塞操作，这会破坏响应式编程的优势。

4. 总结

通过使用 collectList() 操作符，你可以轻松地将 Flux 中的元素收集到 List 中，并将其应用到 Mono 对象中。 这种方法是构建响应式应用程序的常用模式。 记住，理解 Reactor 的操作符和线程模型对于编写高效和可维护的响应式代码至关重要。