我可以将 Flux 作为 ServerResponse 正文的一个字段吗？答案

【问题标题】：Can I have Flux as a field of ServerResponse body?我可以将 Flux 作为 ServerResponse 正文的一个字段吗？
【发布时间】：2021-07-24 10:10:55
【问题描述】：

我是 Spring Reactive Web 的新手，遇到了以下问题。我想创建一个带有接受数字 N 的端点的微服务 A，向微服务 B 发送 N 个请求（它为每个请求返回一个字符串），将字符串包装成对象，将它们组合成 List/Flux（？）并返回包含这些对象的 JSON，例如：

{
  "number": 4,
  "objects": [
    {
       "name": "first"
    },
    {
       "name": "second"
    },
    {
       "name": "third"
    },
    {
       "name": "fourth"
    }
  ]
}

我想为此使用功能端点。所以我尝试了以下方法（尽我所能简化它）：

public class MyObject {
    private String name; // here should be a value received from B
    // ...
}

public class MyResponse {
    private int number;
    private Flux<MyObject> objects; // or List?
    // ...
}

@Component
@RequiredArgsConstructor
public class MyHandler {

    private final MyClient client;

    public Mono<ServerResponse> generate(ServerRequest serverRequest) {
        return serverRequest.bodyToMono(MyRequestBody.class)
                .flatMap(request -> buildServerResponse(HttpStatus.OK, buildResponseBody(request)));
    }

    private Mono<ServerResponse> buildServerResponse(HttpStatus status, Mono<MyResponse> responseBody) {
        return ServerResponse.status(status)
                .contentType(MediaType.APPLICATION_JSON)
                .body(responseBody, MyResponse.class);
    }

    private Mono<MyResponse> buildResponseBody(MyRequestBody request) {
        return Mono.just(MyResponse.builder()
                .number(request.getNumber())
                .objects(getObjects(request.getNumber())
                .build());
    }

    private Flux<MyObject> getObjects(int n) {
        // how to receive n strings from MyClient, make MyObject from each of them and then combine them together to a Flux/List?
    }

public class MyClient {
    public Mono<String> getName() {
        WebClient client = WebClient.builder().baseUrl(getUrl()).build();

        return client.get()
                // ...
                .retrieve()
                .bodyToMono(String.class);
    }

    private String getUrl() {
        // ...
    }
}

所以，如果我在 MyResponse 中使用 Flux，我会收到如下响应：

{
  "number": 4,
  "objects": {
    "prefetch": 2147483647,
    "scanAvailable": true
  }
}

另一方面，如果我尝试使用列表，它似乎在某些时候需要阻塞，并且我收到与它相关的错误。那么，我该怎么做呢？

提前致谢！

更新：如果我使用 collectList().block() 从 Flux 中创建一个列表，我会收到这个：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread <...>

据我对this 问题的回答了解，当我的方法返回Mono/Flux 时，我永远不应该阻止。将block() 调用提取到一个单独的方法，该方法从返回Mono/Flux 的方法调用并没有帮助。如果我在block() 之前使用share()，那么我的请求将永远执行，出于某种我还不明白的原因。

【问题讨论】：

它是一个单声道，包含一个具有属性的对象和其他属性，一个名称为字段名称和值的对象列表，为什么要将其称为通量？
@silentsudo 不太了解你 :(

标签： java spring-webflux project-reactor reactive flux

【解决方案1】：

好的，我做到了。

Flux 作为一个字段不能以预期的方式工作，所以我需要一个List。

public class MyResponse {
    private int number;
    private List<MyObject> objects;
    // ...
}

现在我需要一种方法从多个Mono<String>s 中创建一个List<MyObject>，其中每个MyObject 都有一个String 字段。

问题是我们永远不会摆脱Mono 或Flux，所以我们首先选择Flux<MyObject>。

private Flux<MyObject> getObjects(int n) {
    return Flux.range(0, n) // Flux<Integer>
            .map(i -> myClient.getName()) // Flux<String>
            .map(name -> new MyObject(name)); // Flux<MyObject>
}

然后我们制作 Flux：

private Mono<MyResponse> buildResponseBody(MyRequestBody request) {
    return getObjects(request.getNumber()) // Flux<MyObject>
            .collectList() // Mono<List<MyObject>>
            .map(objects -> MyResponse.builder() // Mono<MyResponse>
                    .number(request.getNumber())
                    .objects(objects)
                    .build()));
}

这样就可以了，因为我们不需要阻止任何东西。

只有当我们想要在某个时候摆脱Mono/Flux 时才会出现问题，比如我们想要一个纯List<MyObject>。但只要我们有 Mono 和/或 Flux 作为输入和输出，我们就可以使用这些类的方法进行所有操作，在每个阶段保留 Mono 或 Flux。

【讨论】：