java - 如何构建避免嵌套 Flux block (Flux>)的响应式架构？

Question

我正在尝试构建一个应用程序 A(如适配器)，它将:

1)接收带有某些 key (JSON格式)的POST请求

2) 它应该以某种方式修改该 key 并向另一个系统 B 创建 POST 请求。

3)应用程序 A 应该解析来自应用程序 B 的响应并修改该响应。

4) 之后我的应用程序 A 应该响应初始 POST 请求。

@RestController
@RequestMapping("/A")
public class Controller {
    @ResponseStatus(HttpStatus.OK)
    @PostMapping(value = "B", consumes = APPLICATION_JSON_VALUE)
    // to return nested Flux is a bad idea here
    private Flux<Flux<Map<String, ResultClass>>> testUpdAcc(@RequestBody Flux<Map<String, SomeClass>> keys) {
        return someMethod(keys);
    }

    // the problem comes here when I will get Flux<Flux<T>> in the return
    public Flux<Flux<Map<String, ResultClass>>> someMethod(Flux<Map<String, SomeClass>> keysFlux) {
        return keysFlux.map(keysMap -> {
                                // do something with keys and create URL
                                // also will batch keys here
                                <...>

                                // for each batch of keys:
                                WebClient.create(hostAndPort)
                                .method(HttpMethod.POST)
                                .uri(url)
                                .body(BodyInserters.fromObject(body))
                                .header(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded")
                                .accept(MediaType.APPLICATION_JSON)
                                .retrieve()
                                .bodyToMono(schema) // response will be parsed into some schema here
                                .retryWhen (// will make a retry mechanism here)

                                // ===== will join all Mono batches into single Flux
                                Flux.concat(...);
                                }
                             );
    }

}

当然，这可以通过不将keyFlux读取为Flux并将其读取为Map来解决。但这应该会让一切变得不那么被动，不是吗？ :)

    @ResponseStatus(HttpStatus.OK)
    @PostMapping(value = "B", consumes = APPLICATION_JSON_VALUE)
    // to return nested Flux is a bad idea here
    private Flux<Map<String, ResultClass>> testUpdAcc(@RequestBody Map<String, SomeClass> keys) {
        return someMethod(keys);
    }

此外，我在返回请求之前的最后一刻尝试使用 block()/blockFirst() ，但出现错误:

block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor...

感谢您的想法!

Answer 1

忘记我的问题 - 我们可以轻松地使用“flatMap”而不是“map”。这将解决 Flux 内部 Flux 的问题。