WebFlux/反应器：使用doOn完成在Flux执行之前检查条件

Question

我已经在用Flux查询一些外部资源了。使用（）。现在我想实现一种乐观锁定：在查询开始执行之前读取一些状态，并检查它是否在查询完成后更新。如果是这样，抛出一些异常以中断http请求处理。

我通过使用doOnComplete实现了这一点：

final AtomicReference<String> initialState = new AtomicReference<>();

return Flux.just("some", "constant", "data")
    .doOnComplete(() -> initialState.set(getState()))
    .concatWith(Flux.using(...)) //actual data query
    .doOnComplete(() -> {if (!initialState.get().equals(getState())) throw new RuntimeException();})
    .concatWithValues("another", "constant", "data")

我的问题：

1. 正确吗？是否保证在通量之前完成第一个λ。使用（），是否保证在之后严格执行第二个doOnCompletelambda
2. 是否存在更优雅的解决方案

Answer 1

第一个doOnComplete将在Flux之后执行。仅（“some”、“constant”、“data”）发出所有元素，在发出后的第二个元素在concatWith中定义，并成功完成。这是可行的，因为两个出版商都有有限数量的元素。

然而，在所提出的方法中，来自特定操作的前置/后置条件在更高级别的操作之外处理。换句话说，属于操作的条件检查泄漏到通量定义中。

建议，将条件检查下推至操作：

var otherElements = Flux.using( // actual data query
        () -> "other",
        x -> {
            var initialState = getState();
            return Flux.just(x).doOnComplete(() ->
                { if (!initialState.equals(getState())) throw new IllegalStateException(); }
            );
        },
        x -> { }
);

Flux.just("some", "constant", "data")
        .concatWith(otherElements)
        .concatWith(Mono.just("another")) // "constant", "data" ...