将流式 Flux 的 WebClient 帖子拆分为 JSON 数组

Question

我正在使用第三方 REST 控制器，它接受 JSON 对象数组并返回单个对象响应。当我使用有限的 Flux 从 WebClient 发布时，代码有效（我假设，因为 Flux 完成）。

但是，当Flux 可能无限时，我该怎么做；

1. 在数组块中发布？
2. 捕获每个 POST 数组的响应？
3. 停止Flux的传输？

这是我的豆子；

public class Car implements Serializable {

    Long id;

    public Car() {}
    public Car(Long id) { this.id = id; }
    public Long getId() {return id; }
    public void setId(Long id) { this.id = id; }
}

这是我假设第三方客户端的样子；

@RestController
public class ThirdPartyServer {

    @PostMapping("/cars")
    public CarResponse doCars(@RequestBody List<Car> cars) {
        System.err.println("Got " + cars);
        return new CarResponse("OK");
    }
}

这是我的代码。当我发布 flux2 时，会发送一个 JSON 数组。但是，当我发布 flux1 时，在第一个 take(5) 之后没有发送任何内容。如何 POST 下一个 5 块？

@Component
public class MyCarClient {

    public void sendCars() {

//      Flux<Car> flux1 = Flux.interval(Duration.ofMillis(250)).map(i -> new Car(i));
        Flux<Car> flux2 = Flux.range(1, 10).map(i -> new Car((long) i));

        WebClient client = WebClient.create("http://localhost:8080");
        client
            .post()
            .uri("/cars")
            .contentType(MediaType.APPLICATION_JSON)
            .body(flux2, Car.class) 
//          .body(flux1.take(5).collectList(), new ParameterizedTypeReference<List<Car>>() {})
            .exchange()
            .subscribe(r -> System.err.println(r.statusCode()));
    }
}

Answer 1

1. 如何在数组块中进行 POST？

使用Flux.window 的一种变体将主通量拆分为窗口通量，然后通过.flatMap 使用窗口通量发送请求

        Flux<Car> flux1 = Flux.interval(Duration.ofMillis(250)).map(i -> new Car(i));

        WebClient client = WebClient.create("http://localhost:8080");
        Disposable disposable = flux1
                // 1
                .window(5)
                .flatMap(windowedFlux -> client
                        .post()
                        .uri("/cars")
                        .contentType(MediaType.APPLICATION_JSON)
                        .body(windowedFlux, Car.class)
                        .exchange()
                        // 2
                        .doOnNext(response -> System.out.println(response.statusCode()))
                        .flatMap(response -> response.bodyToMono(...)))
                .subscribe();

        Thread.sleep(10000);

        // 3
        disposable.dispose();

1. 如何捕获每个 POST 数组的响应？

您可以通过.exchange()之后的运算符来分析响应。

在我提供的示例中，可以在doOnNext 运算符中看到响应，但您可以使用任何对onNext 信号进行操作的运算符，例如map 或handle。

请务必完整阅读响应正文以确保连接返回到池中（请参阅note）。在这里，我使用了.bodyToMono，但任何.body 或.toEntity 方法都可以。

1. 停止 Flux 的传输？

当您使用subscribe 方法时，您可以使用返回的disposable.dispose() 停止流。

或者，您可以从 sendCars() 方法返回 Flux 并将订阅和处置委托给调用者。

请注意，在我提供的示例中，我只是使用Thread.sleep() 来模拟等待。在实际应用中，你应该使用更高级的东西，避免Thread.sleep()