如何将 Mono<List<Object>> 转换为 Flux<Object>？答案

【问题标题】：How to convert a Mono<List<Object>> to a Flux<Object>?如何将 Mono<List<Object>> 转换为 Flux<Object>？
【发布时间】：2020-10-04 10:27:53
【问题描述】：

我刚刚开始使用响应式，所以如果问题没有意义，请纠正我。

我正在创建一个使用 Spring WebFlux + Reactive MongoDB 作为数据库的 rest 控制器，其中包含以下文档结构。

{
    "_id": {
        "$oid": "5ee350839d3d4e34f0790566"
    },
    "customerId": "7777",
    "contacts": [{
            "_id": {
                "$oid": "5ee350839d3d4e34f0790565"
            },
            "name": "Alice",
            "mobileNumbers": "0123456789"
        }, {
            "_id": {
                "$oid": "5ee3508a9d3d4e34f0790567"
            },
            "name": "Tom",
            "mobileNumbers": "1123456789"
        }
    ],
    "_class": "com.demo.contact.model.Customer"
}

此示例文档有一位客户及其联系人。从客户的联系人列表中搜索时，我正在尝试获取 Flux。

public Mono<List<Contact>> searchContacts(String customerId, String searchCriteria) {

    return customerRepository.findById(customerId).map(existingCustomer -> {
        List<Contact> contacts= existingCustomer.getContacts().stream()
                .filter( // some filtering code )
                .sorted(Comparator.comparing(Contact::getName))
                .collect(Collectors.toList());
        return contacts;
    });
}

问题是如何更改这段代码以获得Flux<Object> 而不是Mono<List<Object>>？

【问题讨论】：

您还没有解释将具有可能多个值的 List 减少为单个值的逻辑。
您的列表需要排序吗？如果没有，您可以跳过所有流内容，即findById(customerId).map(Customer::getContacts).flatMapIterable(Function.identity()).filter()

标签： java mongodb reactive-programming spring-webflux

【解决方案1】：

从Mono 到Flux 有几种方法，您将通过经验学习。您将需要在 Java streams 方面做得更好，在 Optional 方面做得更好，当然还有 Reactive API。 Optional 部分很重要，因为您需要了解使用 .map 意味着您仍然有 Optional 结果，因此您也应该包含 .orElse。

您没有指定是否使用响应式 MonogoDb 驱动程序，所以我假设您是。

使用Mono<List<?>> 的一个好方法是使用Mono::flatMapMany 或Mono::flatMapIterable；

public Flux<Contact> searchContacts(String customerId, String searchCriteria) {
    return customerRepository.findMonoById(customerId)
            .map(optionalCustomer -> optionalCustomer
                    .map(existingCustomer -> existingCustomer.getContacts().stream().filter(c -> c == c)
                            .sorted(Comparator.comparing(Contact::getName)).collect(Collectors.toList()))
                    .orElse(Collections.emptyList()))
            .flatMapMany(contacts -> Flux.fromIterable(contacts));
}

或

public Flux<Contact> searchContacts(String customerId, String searchCriteria) {
    return customerRepository.findMonoById(customerId)
            .map(optionalCustomer -> optionalCustomer
                    .map(existingCustomer -> existingCustomer.getContacts().stream().filter(c -> c == c)
                            .sorted(Comparator.comparing(Contact::getName)).collect(Collectors.toList()))
                    .orElse(Collections.emptyList()))
            .flatMapIterable(Function.identity());
}

如果您没有使用响应式驱动程序，那么您可以从一个可迭代对象中创建一个 Flux。

public Flux<Contact> searchContacts(String customerId, String searchCriteria) {
    return Flux.fromIterable(customerRepository.findById(customerId)
            .map(existingCustomer -> existingCustomer.getContacts().stream().filter(c -> c == c)
                    .sorted(Comparator.comparing(Contact::getName)).collect(Collectors.toList()))
            .orElse(Collections.emptyList()));
}

【讨论】：