求助：Micronaut WebSocket服务端会话5秒异常关闭（状态码1011）

问题现象 #

参照Micronaut官方WebSocket示例实现服务端后，使用JavaFX原生WebSocket API或OkHttp客户端连接时，会话均会在5秒左右异常关闭，收到状态码1011，服务端会话也会同步关闭，且会话在发送消息前就提前过期，查阅官方文档未找到有效保活方案。

服务端代码（MyWebsockets.java） #

import io.micronaut.http.MediaType;
import io.micronaut.websocket.WebSocketBroadcaster;
import io.micronaut.websocket.WebSocketSession;
import io.micronaut.websocket.annotation.OnClose;
import io.micronaut.websocket.annotation.OnMessage;
import io.micronaut.websocket.annotation.OnOpen;
import io.micronaut.websocket.annotation.ServerWebSocket;
import org.json.JSONObject;
import org.reactivestreams.Publisher;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@ServerWebSocket("/ws/chat/{tail}/{logs}")
public class MyWebsockets {

    public final WebSocketBroadcaster broadcaster;

    private final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>();

    public MyWebsockets(WebSocketBroadcaster broadcaster) {
        this.broadcaster = broadcaster;
    }

    @OnOpen
    public Publisher<String> onOpen(String tail, String logs, WebSocketSession session) {
        System.out.println("onOpen: " + session.getId());
        sessions.put(session.getId(), session);
        System.out.println("onOpen: " + sessions.size());

        return broadcaster.broadcast(new JSONObject()
                        .put("type", "joined")
                        .put("success", true)
                        .put("logs", logs)
                        .put("tail", tail).toString()
                , MediaType.APPLICATION_JSON_TYPE);
    }

    @OnMessage
    public Publisher<String> onMessage(String tail, String logs, String message, WebSocketSession session) {
        System.out.println("onOpen: " + message);
        return broadcaster.broadcast(String.format("[%s] %s", logs, message), MediaType.APPLICATION_JSON_TYPE);
    }

    @OnClose
    public Publisher<String> onClose(String tail, String logs, WebSocketSession session) {
        System.out.println("[onClose] " + tail + " " + logs + session.getId());
        sessions.remove(session.getId());

        return broadcaster.broadcast(String.format("[%s] Leaving %s!", logs, tail), MediaType.APPLICATION_JSON_TYPE);
    }
}

JavaFX客户端代码 #

public static void main(String [] s){
    new Thread(() -> {
        ws = HttpClient
                .newHttpClient()
                .newWebSocketBuilder().connectTimeout(Duration.of(30, ChronoUnit.SECONDS))
                .buildAsync(URI.create("ws://" + finalUrl +"/ws/chat/room/four"), new WebSocketClient())
                .join();

        ws.sendText("hi",true);
        while (true) {
        }
    }).start();
}

// ... 省略其他代码

private static class WebSocketClient implements WebSocket.Listener {
    Logger logger = Logger.getLogger(WebSocketClient.class.getName());
    
    @Override
    public void onOpen(WebSocket webSocket) {
        WebSocket.Listener.super.onOpen(webSocket);
        WebSockets.IS_ONLINE = Boolean.TRUE;
        logger.info("onOpen: " + webSocket);
    }

    @Override
    public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
        WebSockets.IS_ONLINE = Boolean.TRUE;
        logger.info("onText received " + data);
        System.out.println("onText received " + data);
        return WebSocket.Listener.super.onText(webSocket, data, last);
    }

    @Override
    public CompletionStage<?> onPing(WebSocket webSocket, ByteBuffer message) {
        System.out.println("onPing received " + message);
        return WebSocket.Listener.super.onPing(webSocket, message);
    }

    @Override
    public CompletionStage<?> onPong(WebSocket webSocket, ByteBuffer message) {
        System.out.println("onPong received " + message);
        return WebSocket.Listener.super.onPong(webSocket, message);
    }

    @Override
    public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
        logger.info("onClose:statusCode- " + statusCode);
        WebSockets.IS_ONLINE = Boolean.FALSE;
        return WebSocket.Listener.super.onClose(webSocket, statusCode, reason);
    }

    @Override
    public void onError(WebSocket webSocket, Throwable error) {
        logger.info("Bad day! " + webSocket.toString());
        WebSocket.Listener.super.onError(webSocket, error);
    }
}

问题原因与解决方法 #

1. 响应式流未正确订阅导致会话超时 #

Micronaut的WebSocket @OnOpen/@OnMessage/@OnClose方法返回的Publisher如果没有被框架正确订阅，会触发服务端默认5秒的超时机制，导致会话被强制关闭（状态码1011）。

解决：

• 将方法返回类型改为void，无需返回Publisher；
• 改用broadcaster.broadcastAsync()方法，该方法会自动处理响应式流的订阅逻辑，无需手动返回Publisher。

修改后的@OnOpen示例：

@OnOpen
public void onOpen(String tail, String logs, WebSocketSession session) {
    System.out.println("onOpen: " + session.getId());
    sessions.put(session.getId(), session);
    System.out.println("onOpen: " + sessions.size());

    // 使用broadcastAsync自动处理订阅
    broadcaster.broadcastAsync(new JSONObject()
                    .put("type", "joined")
                    .put("success", true)
                    .put("logs", logs)
                    .put("tail", tail).toString()
            , MediaType.APPLICATION_JSON_TYPE);
}

2. 客户端阻塞线程问题 #

客户端中while(true)会阻塞线程，导致WebSocket事件循环无法处理心跳或服务端消息，间接引发超时。

解决：
替换无限循环为线程等待逻辑：

// 替换while(true)
try {
    Thread.currentThread().join();
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

3. 调整服务端超时配置 #

按需修改服务端WebSocket空闲超时时间，在application.yml中添加：

micronaut:
  server:
    websocket:
      idle-timeout: 30m # 按需设置，示例为30分钟

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