将 Flux 数据填充到 Mono 中的 List 对象

本文旨在指导开发者如何使用 Project Reactor 将 Flux 中的数据收集到 List，并最终将该 List 填充到 Mono 包含的对象的属性中。通过 collectList() 操作符，将 Flux 转换为包含列表的 Mono，再使用 map() 操作符创建并填充目标对象，从而实现数据的转换和整合。

在使用 Project Reactor 进行响应式编程时，经常会遇到需要将 Flux 中的数据收集到一个 List 中，并将该 List 赋值给 Mono 中某个对象的属性。以下是一个详细的教程，说明如何实现这一目标。

核心思路

核心思路是将 Flux 转换为 Mono>，然后再利用 Mono 的 map 操作符创建目标对象并设置其属性。

示例代码

假设我们有以下 Person 类，其中包含一个 List 类型的 items 属性：

import java.util.List;

public class Person {

    private List items;

    public List getItems() {
        return items;
    }

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

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

现在假设我们有一个 service 对象，它提供了一个返回 Flux 的方法 getItems()：

知识画家

AI交互知识生成引擎，一句话生成知识视频、动画和应用

下载

import reactor.core.publisher.Flux;
import java.util.Arrays;

class ItemService {
    public Flux getItems() {
        return Flux.fromIterable(Arrays.asList(
                new Item("Item 1"),
                new Item("Item 2"),
                new Item("Item 3")
        ));
    }
}

以下代码演示了如何将 Flux 中的数据收集到 List，并将其赋值给 Mono 中的 items 属性：

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

import java.util.List;

public class FluxToListInMono {

    public static void main(String[] args) {
        ItemService service = new ItemService(); // 假设的 Service 类

        Mono personMono = createPersonFromReceivedItems(service);

        personMono.subscribe(person -> {
            System.out.println("Person's items: " + person.getItems().toString());
        });
    }

    public static Mono createPersonFromReceivedItems(ItemService service) {
        Flux items = service.getItems(); // 获取 Item 的 Flux

        Mono> collectedItems = items.collectList(); // 将 Flux 转换为 Mono>

        Mono result = collectedItems.map(itemList -> {  // 使用 map 操作符创建 Person 对象并设置 items 属性
            Person p = new Person();
            p.setItems(itemList);
            return p;
        });

        return result;
    }
}

代码解释

1. Flux items = service.getItems();: 从 service 获取 Flux。
2. Mono> collectedItems = items.collectList();: 使用 collectList() 操作符将 Flux 转换为 Mono>。 collectList() 操作符会将 Flux 中的所有元素收集到一个 List 中，并在 Flux 完成后发出该 List。
3. Mono result = collectedItems.map(itemList -> { ... });: 使用 map 操作符，接收 Mono> 发出的 List，创建一个新的 Person 对象，并将 List 设置为 Person 对象的 items 属性。

注意事项

• collectList() 操作符会等待 Flux 完成后才发出 List。 如果 Flux 是无限流，则 collectList() 将永远不会完成，导致程序阻塞。 对于无限流，需要使用其他操作符，例如 window 和 flatMap 来分批处理数据。
• 确保 Person 类中的 setItems() 方法正确地设置了 items 属性。
• 在 subscribe() 方法中，可以根据需要对 Person 对象进行进一步的处理。

总结

通过 collectList() 操作符将 Flux 转换为 Mono>，再利用 Mono 的 map 操作符创建目标对象并设置其属性，是解决将 Flux 数据填充到 Mono 中的 List 对象这一问题的有效方法。 这种方法简洁明了，易于理解和实现。 掌握这种方法，可以更加灵活地处理响应式编程中的数据转换和整合。