通量串联不返回

Posted 2023-02-26

【中文标题】通量串联不返回

【英文标题】：Flux Concatenation does not return

【发布时间】：2017-07-19 21:05:08

【问题描述】：

我正在尝试使用 Spring Boot 2.0、webflux 和 reactiv Mongo 存储库。我有两种变体，首先删除然后将数据添加到集合中。在第一个变体中，线程阻塞直到删除完成，在第二个变体中，数据的添加被连接到删除。

变体 A

@GetMapping("init")
public String init() 
    Random rand = new Random();
    Flux<Power> powers = Flux.range(0, 10000)
            .map(i -> new Power(i,
                    LocalDateTime.now().toEpochSecond(ZoneOffset.of("+1")),
                    rand.nextDouble()));
    powerRepository.deleteAll().block();
    powerRepository.save(powers).blockLast();
    return "ok";

变体 B

@GetMapping("init")
public String init() 
    Random rand = new Random();
    Flux<Power> powers = Flux.range(0, 10000)
            .map(i -> new Power(i,
                    LocalDateTime.now().toEpochSecond(ZoneOffset.of("+1")),
                    rand.nextDouble()));
    powerRepository.deleteAll()
            .concatWith((v) -> powerRepository.save(powers)).blockLast();
    return "ok";

变体 A 返回，变体 B 不返回。有什么区别？结合两个存储库操作的正确方法是什么？

【问题讨论】：

基本上是 Greg 写的。不同之处在于值的发射：deleteAll() 在deleteAll 执行期间终止而不发射值并连接结果。 Concat 在执行父发布者时执行，这可能会导致不良行为。通常，返回反应类型（Mono、Flux）而不是解析类型（String）以保留反应行为。否则，您仍然会阻塞调用线程。 Spring WebFlux 为您处理执行。

【参考方案1】：

如果没有更好的方法，使用 .then 调用链。避免阻塞调用，而是返回 Mono.just("ok")。

public Mono<String> init() 
    return repo.deleteAll()
        .then(() -> repo.save(...))
        .then(() -> Mono.just("ok"));

使端点返回 Mono。

【讨论】：

如果您需要保存实体列表，则返回 Flux 作为响应类型并使用 .thenMany(() -> repo.save(powers)).thenMany(() -> Flux.just("ok"))