我可以将 Flux 作为 ServerResponse 主体的字段吗？

Question

我是 Spring Reactive Web 的新手，我遇到了以下问题。我想创建一个微服务 A，其端点接受数字 N，向微服务 B 发送 N 个请求（returns 每个请求一个字符串），将字符串包装到对象中，将它们组合成一个 List/Flux (?) 和 returns a 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 <...>

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

Answer 1

好的，我做到了。

Flux 因为字段无法按预期方式工作，所以我需要一个 List.

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

现在我需要一种方法来从多个 Mono<String> 中创建一个 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>。但是只要我们有一个Monoand/orFlux作为输入和输出，我们就可以用这些类的方法进行所有操作，保留Mono 或 Flux 在每个阶段。

我可以将 Flux 作为 ServerResponse 主体的字段吗？

Can I have Flux as a field of ServerResponse body?

java

flux

project-reactor

reactive

spring-webflux