Spring Webflux SSE模式下Flux执行完成后触发cancel()事件问题

我之前在使用早期Spring Webflux版本的时候也碰到过类似的情况，结合你的场景（Spring Boot 2.0.0.M7 + Webflux + Thymeleaf Reactive），这个问题大概率和版本本身的bug或者集成逻辑的小问题有关，下面给你拆解原因和解决方案：

可能的原因分析 #

• 早期Webflux版本的SSE生命周期bug：2.0.0.M7是Spring Boot 2.0系列的早期里程碑版本，当时Webflux的SSE响应处理逻辑还不完善，存在响应完成后错误触发cancel信号的情况，后续正式版已经修复了这类问题。
• Thymeleaf Reactive集成的资源管理问题：Thymeleaf的Reactive模块在早期版本中，可能对SSE类型的响应存在额外的资源清理逻辑，导致在Flux触发onComplete后，主动调用了cancel方法。
• 客户端/网关的连接干扰：如果你的微服务前面有网关或者代理，某些网关在收到SSE的结束信号后会主动断开连接，进而触发Flux的cancel，但你提到其他场景不会出现，这个可能性相对较低。

具体解决方案 #

1. 优先升级Spring Boot版本 #

这是最直接有效的解决办法，把Spring Boot从2.0.0.M7升级到2.0.x的正式版（比如2.0.9.RELEASE），或者直接升级到更稳定的2.1+版本。官方在后续版本中修复了大量Webflux和SSE相关的bug，包括响应生命周期的管理问题。

2. 规范SSE端点的实现方式 #

确保你的端点返回的是ServerSentEvent类型，而不是直接返回原始的Flux，这样Webflux能更准确地处理SSE的生命周期：

@GetMapping(value = "/your-sse-endpoint", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> sseEndpoint() {
    return Flux.range(1, 5)
            .delayElements(Duration.ofSeconds(1))
            .doOnComplete(() -> System.out.println("Flux执行完成，触发onComplete"))
            .doOnCancel(() -> System.out.println("Flux被取消，触发cancel"))
            .map(count -> ServerSentEvent.<String>builder()
                    .id(String.valueOf(count))
                    .data("当前计数：" + count)
                    .build());
}

3. 排查Thymeleaf Reactive的配置 #

检查你的Thymeleaf配置，确保没有自定义的拦截器或者过滤器干扰SSE响应的处理。比如，避免在全局拦截器中对text/event-stream类型的响应做额外的处理，防止意外触发连接关闭。

4. 增加日志定位问题根源 #

如果暂时无法升级版本，可以在Flux链中添加更详细的日志，追踪cancel信号的来源：

return yourOriginalFlux
        .doOnSubscribe(subscription -> System.out.println("Flux被订阅，订阅ID：" + subscription))
        .doOnComplete(() -> System.out.println("onComplete事件触发"))
        .doOnCancel(() -> {
            StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
            System.out.println("cancel事件触发，调用栈：");
            for (StackTraceElement element : stackTrace) {
                System.out.println("\t" + element);
            }
        });

通过调用栈可以清楚看到cancel是从Webflux内部、Thymeleaf还是客户端发起的，进而针对性解决。

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