Spring Webflux Websocket在onErrorResume逻辑可以完成之前关闭

基础概念

Spring WebFlux 是 Spring 框架的一个子项目，它提供了响应式编程模型，支持非阻塞的异步操作。WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。Spring WebFlux 提供了对 WebSocket 的支持，使得开发者可以轻松地在 Web 应用程序中实现实时通信。

类型

Spring WebFlux 中的 WebSocket 处理主要涉及以下几种类型：

简单 WebSocket 处理：使用 WebSocketHandler 接口处理 WebSocket 连接。
基于注解的处理：使用 @Controller 和 @MessageMapping 注解处理 WebSocket 消息。

应用场景

WebSocket 适用于需要实时通信的应用场景，例如：

在线聊天应用
实时数据更新（如股票行情、天气预报）
在线游戏

问题分析

在 Spring WebFlux 中，onErrorResume 是一个用于处理异常并恢复操作的函数。如果在 onErrorResume 逻辑完成之前 WebSocket 连接被关闭，可能会导致以下问题：

异常处理不完整：onErrorResume 逻辑没有机会执行，导致异常没有被正确处理。
资源泄漏：WebSocket 连接关闭时，可能没有正确释放相关资源。

原因

WebSocket 连接在 onErrorResume 逻辑完成之前关闭的原因可能有：

客户端主动关闭连接：客户端可能在服务器端处理异常之前主动关闭了连接。
网络问题：网络不稳定或中断可能导致连接提前关闭。
服务器端逻辑错误：服务器端的某些逻辑可能导致连接提前关闭。

解决方法

为了确保 onErrorResume 逻辑能够完整执行，可以采取以下措施：

检查连接状态：在 onErrorResume 逻辑中，首先检查连接是否仍然打开。
使用 Mono 和 Flux 的错误处理机制：利用响应式编程中的错误处理机制，确保异常能够被捕获和处理。
资源管理：确保在连接关闭时正确释放资源。

以下是一个示例代码，展示了如何在 onErrorResume 中处理异常并确保连接关闭时释放资源：

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Mono;

public class MyWebSocketHandler implements WebSocketHandler {

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        return session.send(
            session.receive()
                .map(message -> {
                    // 处理接收到的消息
                    return message.getPayloadAsText();
                })
                .doOnError(throwable -> {
                    // 处理异常
                    System.err.println("Error processing message: " + throwable.getMessage());
                })
                .onErrorResume(throwable -> {
                    // 恢复操作
                    if (session.isOpen()) {
                        return session.send(
                            session.textMessage("Error occurred: " + throwable.getMessage())
                        );
                    } else {
                        return Mono.empty();
                    }
                })
        ).andWriteWith(Mono.empty())
          .doOnTerminate(() -> {
              // 连接关闭时释放资源
              System.out.println("WebSocket connection closed");
          });
    }
}