Spring Boot WebFlux 实现数据库实时推送

本文详解如何在 Spring WebFlux 中使用 `Flux.interval` 定期触发异步数据库查询，通过 `flatMap` 正确组合 `Mono`（如查询最新文档）与 `Flux`，实现每秒拉取并发布数据库中最新记录的描述字段。

在响应式编程中，不能将阻塞式思维套用于异步流操作。例如，直接在 map() 中调用返回 Mono 的仓库方法（如 findLastDocument()），会导致编译错误或运行时异常——因为 map 仅支持同步转换，而 Mono 是一个异步容器，其值需通过响应式链式操作（如 flatMap）进行“解包”与后续处理。

正确做法是：使用 flatMap 将每个定时信号（由 Flux.interval 发出）映射为一个新的异步查询流。flatMap 会订阅每个 Mono 并将其扁平化为 Flux 元素，从而保持整个流的响应式特性。

以下是修正后的控制器代码：

@RestController
public class WebFluxController {

    @Autowired
    private ReactiveDocumentRepository reactiveDocumentRepository;

    @CrossOrigin
    @GetMapping(value = "/documents", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux getDocuments() {
        return Flux.interval(Duration.ofSeconds(1))
                .onBackpressureLatest() // 防止下游消费慢导致事件积压
                .flatMap(x -> reactiveDocumentRepository.findLastDocument()
                        .map(document -> "document-" + document.getDescription())
                        .defaultIfEmpty("document-NO_DATA")); // 处理查无结果场景
    }
}

? 关键说明：

• produces = MediaType.TEXT_EVENT_STREAM_VALUE 显式声明返回 SSE（Server-Sent Events）格式，便于浏览器或客户端以流式方式接收持续更新；
• onBackpressureLatest() 可选但推荐，避免因数据库响应延迟或客户端读取缓慢造成内存溢出；
• defaultIfEmpty() 提供兜底值，防止 Mono.empty() 导致流中断（flatMap 遇到空 Mono 会跳过该次发射）；
• findLastDocument() 必须是真正响应式的实现（如基于 Spring Data R2DBC 或 Reactive MongoDB），否则将破坏响应式契约。

⚠️ 注意事项：

• 频繁轮询数据库（如每秒一次）并非高并发场景下的最优实践。生产环境中建议结合变更通知机制（如 PostgreSQL 的 LISTEN/NOTIFY、MongoDB Change Streams）实现真正的事件驱动推送；
• 若数据库查询耗时较长，应考虑添加超时控制：

  .timeout(Duration.ofSeconds(2), Mono.just("document-TIMEOUT"))

• 确保 ReactiveDocumentRepository 接口正确定义了 findLastDocument() 方法，例如：

  Mono findLastDocument();

综上，flatMap 是连接定时触发器与异步数据源的核心桥梁。掌握它与 map、switchMap、concatMap 的语义差异，是构建健壮 WebFlux 流的关键一步。