Spring Reactor：上游发布器重启动时如何清除流中在处理元素？

这个问题其实是Reactor流生命周期管理和分布式状态一致性的典型场景，结合你用的Spring Boot 2.2.4和Reactor 3.3.2版本，我给你几个可行的解决方案，从流控制到数据库层面的优化都有：

方案1：通过流生命周期管理，重启前终止旧流 #

你的核心问题之一是旧流还在处理实体时，新的流已经启动并读取了相同的数据。我们可以通过Disposable来管理当前流的订阅，在重启新流前先彻底取消旧流的所有操作，确保正在处理的缓冲和重试任务都被终止。

代码示例： #

// 用原子引用保存当前流的Disposable，方便随时取消
AtomicReference<Disposable> currentProcessingStream = new AtomicReference<>();

// 封装完整的处理流创建逻辑
Supplier<Flux<Void>> buildProcessingFlow = () -> 
    myMongoRepository.executeLargeQuery()
        .filter(entity -> !entity.isProcessed())
        .buffer(10)
        .concatMap(bufferedList -> 
            Flux.fromIterable(bufferedList)
                .flatMap(this::makeExternalCall)
                .then()
        );

// 启动或重启流的方法
private void restartProcessingStream() {
    // 先取消旧流：所有正在处理的flatMap、重试任务都会被终止
    Optional.ofNullable(currentProcessingStream.getAndSet(null))
        .ifPresent(Disposable::dispose);
    
    // 订阅新流，并保存Disposable
    Disposable newStream = buildProcessingFlow.get()
        .onErrorResume(ex -> {
            // 这里根据实际触发重启的条件处理，比如Mongo查询超时、连接异常
            restartProcessingStream();
            return Mono.empty();
        })
        .subscribe();
    
    currentProcessingStream.set(newStream);
}

这样一来，当流需要重启时，旧的makeExternalCall重试任务会被立即取消，不会继续执行标记processed的操作，新流启动时读取的实体都是真正未处理的，避免了重复实例。

方案2：给实体添加处理锁，从数据库层面避免重复 #

流控制能解决大部分场景，但如果遇到极端情况（比如旧流取消前已经触发了外部调用），还是可能出现重复。这时可以给实体加一个处理中状态，用Mongo的原子更新来实现乐观锁，确保同一时间只有一个处理流程能操作实体。

步骤与代码示例： #

1. 给Entity新增一个processingStatus字段（比如：0=未处理，1=处理中，2=已处理）
2. 查询时只读取processingStatus=0的实体
3. 在makeExternalCall开始前，先尝试将实体锁定为processingStatus=1（原子操作）
4. 锁定成功才执行外部调用，失败则直接跳过（说明已有其他流程在处理）

// 第一步：查询只获取未处理且未被锁定的实体
Flux<Entity> entitiesFromMongoDb = myMongoRepository.findByProcessingStatus(0);

// 第二步：修改makeExternalCall的逻辑，加入锁机制
private Mono<Void> makeExternalCall(Entity entity) {
    // 原子更新：只有当processingStatus还是0时，才设置为1，返回更新后的实体
    return myMongoRepository.lockEntityById(entity.getId())
        .filter(Optional::isPresent) // 过滤掉锁定失败的情况
        .flatMap(lockedEntity -> 
            // 执行外部调用
            externalRemoteService.call(lockedEntity.get())
                // 成功后标记为已处理
                .then(myMongoRepository.markEntityAsProcessed(lockedEntity.get().getId()))
                // 失败则解锁，允许后续重试或其他流程处理
                .onErrorResume(ex -> 
                    myMongoRepository.unlockEntityById(lockedEntity.get().getId())
                        .then(Mono.error(ex))
                )
        )
        .then(); // 锁定失败的话直接完成，不做任何操作
}

这个方案的优势是即使流意外重启，数据库层面的锁会阻止重复处理，而且原子更新的性能开销远小于每次查询状态，不会显著增加数据库负载。

方案3：优化Mongo查询的重试策略，减少不必要的流重启 #

你提到executeLargeQuery()会触发重启动，先排查下重启的原因：是查询超时？还是订阅被意外取消？可以给Mongo查询添加精准的重试策略，只在必要时重试，并且重试前清理旧流。

代码示例： #

Flux<Entity> entitiesFromMongoDb = myMongoRepository.executeLargeQuery()
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(5))
        // 只在特定异常时重试，比如Mongo超时异常
        .filter(ex -> ex instanceof MongoTimeoutException)
        // 重试前取消旧的处理流
        .doBeforeRetry(retrySignal -> 
            Optional.ofNullable(currentProcessingStream.get()).ifPresent(Disposable::dispose)
        ));

总结推荐 #

如果流重启是业务上不可避免的场景，方案1+方案2是最优组合：既通过流生命周期管理终止旧的处理任务，又通过数据库锁做最后一道保障，完全避免重复处理。如果可以减少流重启的情况，先优化Mongo查询的重试策略，再配合乐观锁即可。

内容的提问来源于stack exchange，提问作者Johan