Spring Boot WebFlux 实现数据库持续轮询获取最新记录

本文详解如何在 spring webflux 中正确构建响应式流，通过定时轮询（如每秒一次）异步查询数据库并发布最新记录，重点解决 `mono` 与 `flux` 的嵌套组合问题，避免阻塞和类型不匹配错误。

在响应式编程中，不能将阻塞式思维直接迁移到 Mono/Flux 上。例如，reactiveDocumentRepository.findLastDocument() 返回的是 Mono，它代表一个异步、延迟计算、可能为空的单值响应式序列；而 Flux.interval(...) 生成的是一个周期性触发的 Flux。若错误地使用 .map() 尝试将 Mono “塞进”字符串拼接（如 "document-" + mono.map(...)），会导致编译失败或运行时返回 Mono.toString() 这类无意义字符串——因为 map 是同步转换函数，无法“等待” Mono 发出数据。

✅ 正确做法是使用 flatMap：它专为将每个上游元素（这里是 Long）映射为一个新的响应式序列（Mono）并自动展平（flatten）为 Flux 而设计。flatMap 会订阅每个 Mono，等待其发出数据后合并到最终流中，真正实现“每秒查一次库、取最新文档描述、发一条消息”的语义。

以下是修正后的完整控制器示例：

@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))
                .onBackpressureDrop() // 防止下游消费过慢导致内存溢出
                .flatMap(x -> reactiveDocumentRepository.findLastDocument()
                        .map(document -> "document-" + document.getDescription())
                        .defaultIfEmpty("document-none")); // 若无记录，提供兜底值
    }
}

? 关键要点说明：

魔珐星云

无需昂贵GPU，一键解锁超写实/二次元等多风格3D数字人，跨端适配千万级并发的具身智能平台。

下载

• produces = MediaType.TEXT_EVENT_STREAM_VALUE：显式声明返回 SSE（Server-Sent Events）格式，便于前端用 EventSource 流式接收，提升实时性体验；
• onBackpressureDrop()：当客户端处理速度慢于服务端推送频率时，自动丢弃来不及发送的数据，防止 OutOfMemoryError；
• defaultIfEmpty()：处理数据库暂无数据的边界情况，避免流因 Mono 为空而终止（flatMap 中若 Mono 为空，则不发射任何元素）；
• 绝不使用 block() 或 toFuture().get()：这将彻底破坏响应式非阻塞特性，导致线程池耗尽和性能崩溃。

⚠️ 注意事项：

• 数据库查询本身必须是真正的响应式操作（如基于 R2DBC 或 Spring Data MongoDB Reactive），否则 findLastDocument() 内部若含 JDBC 阻塞调用，整个链路仍会退化为伪响应式；
• 高频轮询（如每秒 1 次）对数据库有一定压力，生产环境建议结合变更通知机制（如 PostgreSQL 的 LISTEN/NOTIFY、MongoDB Change Streams）替代轮询；
• 若需保证严格顺序或去重，应在 flatMap 后添加 .distinctUntilChanged() 或结合时间戳做业务级幂等判断。

综上，WebFlux 中的持续数据流不是“循环+sleep+查询”的命令式模拟，而是通过 Flux.interval 触发信号、flatMap 编排异步查询、map/defaultIfEmpty 等操作符加工结果的声明式组合。掌握 map（同步转换）与 flatMap（异步序列展开）的本质区别，是写出健壮响应式服务的核心前提。