Spring Webflux SSE场景下如何检测客户端意外离线？

针对你遇到的Spring Webflux + Netty下SSE客户端意外断网（网卡禁用、信号丢失）无法被服务器检测的问题，可以通过以下方式结合解决：

1. 配置Netty应用层空闲检测 #

利用Netty的IdleStateHandler检测客户端的读空闲状态（即客户端长时间无数据发送），触发超时后主动关闭连接，让Flux链正常终止。

在Spring Boot中添加Netty自定义配置类：

import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.boot.web.embedded.netty.NettyServerCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class NettyIdleConfig {

    @Bean
    public NettyServerCustomizer idleDetectionCustomizer() {
        return httpServer -> httpServer.tcpConfiguration(tcpServer ->
                tcpServer.doOnConnection(connection -> {
                    // 检测10秒读空闲（客户端无数据发送）
                    connection.channel().pipeline().addLast(new IdleStateHandler(10, 0, 0));
                    // 自定义处理器，触发空闲事件时关闭连接
                    connection.channel().pipeline().addLast(new ChannelDuplexHandler() {
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                            if (evt instanceof IdleStateEvent) {
                                IdleStateEvent idleEvent = (IdleStateEvent) evt;
                                if (idleEvent.state() == IdleState.READER_IDLE) {
                                    System.out.println("检测到客户端读空闲，关闭连接");
                                    ctx.close();
                                }
                            }
                            super.userEventTriggered(ctx, evt);
                        }
                    });
                })
        );
    }
}

2. 优化SSE端点的错误处理 #

在Flux链中添加错误处理逻辑，当连接断开导致写操作失败时，主动终止Flux：

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import reactor.netty.http.server.ServerSentEvent;
import java.time.Duration;

@RestController
@RequestMapping("/")
public class SSEController {

    @GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<Object>> streamEvents() {
        System.out.println("客户端已连接");

        return Flux.interval(Duration.ofSeconds(5))
                .map(second -> {
                    System.out.println("发送心跳...");
                    return ServerSentEvent.<Object>builder()
                            .event("heartbeat")
                            .build();
                })
                .doFinally(signalType -> System.out.println("客户端已断开连接"))
                .onErrorResume(error -> {
                    System.out.println("连接异常：" + error.getMessage());
                    return Flux.empty();
                })
                .subscribeOn(Schedulers.parallel());
    }
}

测试方法 #

• 本地浏览器访问http://localhost:8080，然后禁用网卡/断开Wi-Fi，等待10秒左右，观察服务器日志是否打印检测断开的相关信息。
• 使用curl http://localhost:8080命令建立连接，断开网络后等待超时，查看日志输出。
• 编写Java客户端模拟连接，手动切断网络，验证服务器的检测逻辑。

原理说明 #

客户端意外断网时，TCP连接不会主动发送FIN/RST包，服务器默认无法感知。通过Netty的IdleStateHandler在应用层检测客户端空闲状态，比系统默认的TCP Keepalive（通常默认2小时超时）更灵敏，能快速识别死连接并关闭，进而触发Flux的doFinally逻辑。

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