如何在Flux上游无可用元素时重复发送最后一个元素？

嘿，这个场景我之前刚好处理过，完全匹配你说的「反向背压」需求——订阅者处理速度比发布者快，得持续拿到最新的（或者说最后一个）元素，而不是干等发布者生产新内容。咱们一步步来实现这个逻辑：

核心思路拆解 #

你需要的是三个关键行为的组合：

• 跟踪并缓存最后一个元素：每当发布者推送新元素时，立刻更新这个缓存值
• 持续向订阅者推送元素：当发布者暂时没有新元素产出时，重复推送缓存的最后一个元素
• 向上游请求无限需求：让发布者不用等订阅者的请求信号，一有新元素就立刻推送（和onBackpressureDrop的请求模式一致）

具体代码实现 #

我们可以用Reactor的原生操作符组合来实现，不需要自定义发布者：

import reactor.core.publisher.Flux;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
import java.time.Duration;

// 模拟一个生产速度慢的原始发布者
Flux<String> slowPublisher = Flux.just("A", "B", "C")
        .delayElements(Duration.ofSeconds(2));

// 1. 用线程安全的原子引用保存最后一个元素
AtomicReference<String> lastElementRef = new AtomicReference<>();

// 2. 合并两个流：原始发布者的元素流 + 持续重复最后元素的流
Flux<String> repeatingLastFlux = Flux.merge(
        // 第一个流：处理原始发布者的元素，同时更新缓存的最后元素
        slowPublisher.doOnNext(lastElementRef::set),
        // 第二个流：持续发射最后一个元素（确保有初始元素后才推送）
        Flux.interval(Duration.ZERO)
                .map(ignored -> lastElementRef.get())
                .filter(Objects::nonNull)
)
// 3. 向上游请求无限需求，让发布者持续生产，不用考虑订阅者的处理速度
.onBackpressureDrop();

// 测试订阅：模拟处理速度快的订阅者
repeatingLastFlux.subscribe(
        elem -> System.out.println("Received: " + elem),
        error -> System.err.println("Error: " + error),
        () -> System.out.println("Completed")
);

代码细节解释 #

• AtomicReference：线程安全地存储最后一个元素，避免多线程环境下的竞态问题
• Flux.merge：把原始流和重复流合并，确保原始发布者有新元素时立刻推送，空闲时则持续推送缓存的最后元素
• Flux.interval(Duration.ZERO)：创建无间隔的触发流，你可以根据需求调整间隔（比如Duration.ofMillis(100)）来控制重复推送的频率
• onBackpressureDrop()：告诉上游我们能处理无限量的元素，不需要背压机制，确保发布者一有新元素就推送

注意事项 #

• 如果原始发布者一开始没有元素，filter(Objects::nonNull)会避免推送null，直到第一个元素产出
• 重复频率可以根据业务需求灵活调整，不需要固定为无间隔
• 整个方案是完全非阻塞的，符合Reactor响应式编程的核心原则

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