跳到主要内容

Java 服务端实现

Java 提供了标准的 WebSocket API（JSR 356 / Jakarta WebSocket），支持创建企业级 WebSocket 应用。本章介绍如何使用 Jakarta WebSocket 和 Spring Boot 实现 WebSocket 服务端。

Jakarta WebSocket API​

Jakarta WebSocket 是 Java 平台的标准 WebSocket API，定义在 JSR 356 规范中。从 Jakarta EE 9 开始，包名从 javax.websocket 改为 jakarta.websocket。

依赖配置​

Maven 依赖（Jakarta EE 9+）：

<dependency>
    <groupId>jakarta.websocket</groupId>
    <artifactId>jakarta.websocket-api</artifactId>
    <version>2.1.0</version>
    <scope>provided</scope>
</dependency>

Maven 依赖（Java EE 7/8）：

<dependency>
    <groupId>javax.websocket</groupId>
    <artifactId>javax.websocket-api</artifactId>
    <version>1.1</version>
    <scope>provided</scope>
</dependency>

核心包结构​

Jakarta WebSocket 包含两个主要包：

  • jakarta.websocket：客户端和服务端通用的注解、类和接口
  • jakarta.websocket.server：创建和配置服务端端点的注解、类和接口

注解式端点​

注解式端点是创建 WebSocket 服务最简单的方式，通过在类和方法上添加注解来定义端点行为。

基本端点​

import jakarta.websocket.OnMessage;
import jakarta.websocket.server.ServerEndpoint;
import jakarta.websocket.Session;
import java.io.IOException;

@ServerEndpoint("/echo")
public class EchoEndpoint {
    
    @OnMessage
    public void onMessage(Session session, String message) throws IOException {
        // 接收消息并回显
        session.getBasicRemote().sendText("Echo: " + message);
    }
}

@ServerEndpoint 注解指定端点的 URL 路径，@OnMessage 注解标记处理消息的方法。

生命周期注解​

WebSocket 端点有四个生命周期事件，对应四个注解：

注解事件说明
@OnOpen连接打开客户端连接建立时调用
@OnMessage消息接收收到客户端消息时调用
@OnClose连接关闭连接关闭时调用
@OnError错误发生发生错误时调用
import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import java.io.IOException;

@ServerEndpoint("/chat")
public class ChatEndpoint {
    
    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        System.out.println("客户端连接: " + session.getId());
        
        // 获取握手请求信息
        String origin = (String) config.getUserProperties()
            .get("jakarta.websocket.endpoint.http.request.origin");
        System.out.println("Origin: " + origin);
    }
    
    @OnMessage
    public void onMessage(Session session, String message) throws IOException {
        System.out.println("收到消息: " + message);
        session.getBasicRemote().sendText("收到: " + message);
    }
    
    @OnClose
    public void onClose(Session session, CloseReason reason) {
        System.out.println("连接关闭: " + session.getId());
        System.out.println("关闭码: " + reason.getCloseCode().getCode());
        System.out.println("关闭原因: " + reason.getReasonPhrase());
    }
    
    @OnError
    public void onError(Session session, Throwable throwable) {
        System.err.println("发生错误: " + throwable.getMessage());
    }
}

发送消息​

通过 Session 对象的 getBasicRemote() 或 getAsyncRemote() 方法获取远程端点，然后发送消息：

同步发送：

@OnMessage
public void onMessage(Session session, String message) throws IOException {
    // 获取基本远程端点（同步）
    RemoteEndpoint.Basic remote = session.getBasicRemote();
    
    // 发送文本消息
    remote.sendText("文本消息");
    
    // 发送二进制消息
    ByteBuffer buffer = ByteBuffer.wrap(new byte[]{1, 2, 3, 4});
    remote.sendBinary(buffer);
    
    // 发送 Ping
    remote.sendPing(ByteBuffer.wrap(new byte[]{1, 2, 3}));
    
    // 发送 Pong
    remote.sendPong(ByteBuffer.wrap(new byte[]{1, 2, 3}));
}

异步发送：

@OnMessage
public void onMessage(Session session, String message) {
    // 获取异步远程端点
    RemoteEndpoint.Async asyncRemote = session.getAsyncRemote();
    
    // 异步发送文本
    asyncRemote.sendText("异步消息");
    
    // 异步发送二进制
    asyncRemote.sendBinary(ByteBuffer.wrap(new byte[]{1, 2, 3}));
    
    // 带回调的异步发送
    asyncRemote.sendText("消息", new SendHandler() {
        @Override
        public void onResult(SendResult result) {
            if (result.isOK()) {
                System.out.println("发送成功");
            } else {
                System.err.println("发送失败: " + result.getException());
            }
        }
    });
}

接收不同类型消息​

一个端点可以定义多个 @OnMessage 方法处理不同类型的消息：

@ServerEndpoint("/messages")
public class MessageTypeEndpoint {
    
    // 处理文本消息
    @OnMessage
    public void onTextMessage(Session session, String message) {
        System.out.println("文本消息: " + message);
    }
    
    // 处理二进制消息
    @OnMessage
    public void onBinaryMessage(Session session, ByteBuffer buffer) {
        System.out.println("二进制消息: " + buffer.remaining() + " 字节");
    }
    
    // 处理 Pong 消息
    @OnMessage
    public void onPongMessage(Session session, PongMessage pong) {
        System.out.println("Pong 消息");
    }
}

每个端点每种消息类型最多只能有一个处理方法。

编程式端点​

编程式端点通过继承 Endpoint 类实现，提供更细粒度的控制。

基本实现​

import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpointConfig;
import java.io.IOException;
import java.nio.ByteBuffer;

public class ProgrammaticEndpoint extends Endpoint {
    
    @Override
    public void onOpen(Session session, EndpointConfig config) {
        System.out.println("连接打开: " + session.getId());
        
        // 添加消息处理器
        session.addMessageHandler(new MessageHandler.Whole<String>() {
            @Override
            public void onMessage(String message) {
                System.out.println("收到文本消息: " + message);
                try {
                    session.getBasicRemote().sendText("Echo: " + message);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
        
        // 添加二进制消息处理器
        session.addMessageHandler(new MessageHandler.Whole<ByteBuffer>() {
            @Override
            public void onMessage(ByteBuffer message) {
                System.out.println("收到二进制消息: " + message.remaining() + " 字节");
            }
        });
    }
    
    @Override
    public void onClose(Session session, CloseReason closeReason) {
        System.out.println("连接关闭: " + closeReason);
    }
    
    @Override
    public void onError(Session session, Throwable thr) {
        System.err.println("发生错误: " + thr.getMessage());
    }
}

部署编程式端点​

编程式端点需要通过 ServerEndpointConfig 配置部署：

import jakarta.websocket.server.ServerEndpointConfig;
import jakarta.websocket.Endpoint;

// 创建配置
ServerEndpointConfig config = ServerEndpointConfig.Builder
    .create(ProgrammaticEndpoint.class, "/programmatic")
    .build();

在 Servlet 容器中，通常会通过 ServerApplicationConfig 接口自动部署：

import jakarta.websocket.server.ServerApplicationConfig;
import jakarta.websocket.server.ServerEndpointConfig;
import java.util.HashSet;
import java.util.Set;

public class WebSocketConfig implements ServerApplicationConfig {
    
    @Override
    public Set<ServerEndpointConfig> getEndpointConfigs(Set<Class<? extends Endpoint>> endpointClasses) {
        Set<ServerEndpointConfig> configs = new HashSet<>();
        
        for (Class<? extends Endpoint> endpointClass : endpointClasses) {
            if (endpointClass == ProgrammaticEndpoint.class) {
                configs.add(ServerEndpointConfig.Builder
                    .create(endpointClass, "/programmatic")
                    .build());
            }
        }
        
        return configs;
    }
    
    @Override
    public Set<Class<?>> getAnnotatedEndpointClasses(Set<Class<?>> scanned) {
        // 返回需要部署的注解端点类
        Set<Class<?>> endpoints = new HashSet<>();
        for (Class<?> clazz : scanned) {
            if (clazz.getPackage().getName().equals("com.example.endpoints")) {
                endpoints.add(clazz);
            }
        }
        return endpoints;
    }
}

Session 管理​

Session 对象​

Session 对象代表客户端与服务端之间的 WebSocket 会话：

@ServerEndpoint("/session")
public class SessionEndpoint {
    
    @OnOpen
    public void onOpen(Session session) {
        // 获取会话 ID
        String id = session.getId();
        
        // 检查是否开放
        boolean isOpen = session.isOpen();
        
        // 获取最大空闲超时（毫秒）
        long timeout = session.getMaxIdleTimeout();
        
        // 设置超时
        session.setMaxIdleTimeout(60000L);
        
        // 获取最大消息缓冲区大小
        int maxBufferSize = session.getMaxTextMessageBufferSize();
        
        // 设置缓冲区大小
        session.setMaxTextMessageBufferSize(65536);
    }
}

存储用户数据​

Session 提供了用户属性映射，用于存储会话相关数据：

@ServerEndpoint("/user")
public class UserEndpoint {
    
    @OnOpen
    public void onOpen(Session session) {
        // 存储用户数据
        session.getUserProperties().put("username", "匿名用户");
        session.getUserProperties().put("joinTime", System.currentTimeMillis());
    }
    
    @OnMessage
    public void onMessage(Session session, String message) {
        // 获取用户数据
        String username = (String) session.getUserProperties().get("username");
        Long joinTime = (Long) session.getUserProperties().get("joinTime");
        
        System.out.println(username + " 说: " + message);
        System.out.println("已在线: " + (System.currentTimeMillis() - joinTime) / 1000 + " 秒");
        
        // 更新用户名
        if (message.startsWith("/name ")) {
            String newName = message.substring(6);
            session.getUserProperties().put("username", newName);
            sendText(session, "用户名已更新为: " + newName);
        }
    }
    
    private void sendText(Session session, String text) {
        try {
            session.getBasicRemote().sendText(text);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

广播消息​

获取所有打开的会话并广播消息：

import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Set;

@ServerEndpoint("/broadcast")
public class BroadcastEndpoint {
    
    // 存储所有连接（线程安全）
    private static final Set<Session> sessions = 
        java.util.Collections.newSetFromMap(
            new java.util.concurrent.ConcurrentHashMap<>()
        );
    
    @OnOpen
    public void onOpen(Session session) {
        sessions.add(session);
        broadcast("用户加入，当前在线: " + sessions.size());
    }
    
    @OnClose
    public void onClose(Session session) {
        sessions.remove(session);
        broadcast("用户离开，当前在线: " + sessions.size());
    }
    
    @OnMessage
    public void onMessage(Session session, String message) {
        broadcast(session.getUserProperties().get("username") + ": " + message);
    }
    
    private void broadcast(String message) {
        for (Session session : sessions) {
            if (session.isOpen()) {
                try {
                    session.getBasicRemote().sendText(message);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

也可以通过 Session.getOpenSessions() 获取当前端点的所有打开会话：

@OnMessage
public void onMessage(Session session, String message) {
    // 获取当前端点的所有打开会话
    Set<Session> openSessions = session.getOpenSessions();
    
    for (Session s : openSessions) {
        if (s.isOpen() && s != session) {
            try {
                s.getBasicRemote().sendText(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

路径参数​

WebSocket 端点支持 URI 模板参数，用于从 URL 中提取动态值。

基本用法​

import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import jakarta.websocket.*;

@ServerEndpoint("/chat/{room}")
public class RoomEndpoint {
    
    @OnOpen
    public void onOpen(Session session, @PathParam("room") String room) {
        System.out.println("加入房间: " + room);
        session.getUserProperties().put("room", room);
    }
    
    @OnMessage
    public void onMessage(Session session, String message, 
                          @PathParam("room") String room) {
        System.out.println("房间 " + room + " 收到消息: " + message);
        
        // 只广播给同一房间的用户
        for (Session s : session.getOpenSessions()) {
            String sRoom = (String) s.getUserProperties().get("room");
            if (room.equals(sRoom) && s.isOpen()) {
                try {
                    s.getBasicRemote().sendText(message);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

多个路径参数​

@ServerEndpoint("/chat/{room}/{username}")
public class MultiParamEndpoint {
    
    @OnOpen
    public void onOpen(Session session, 
                       @PathParam("room") String room,
                       @PathParam("username") String username) {
        System.out.println(username + " 加入房间 " + room);
    }
    
    @OnMessage
    public void onMessage(Session session, String message,
                          @PathParam("room") String room,
                          @PathParam("username") String username) {
        String formatted = String.format("[%s] %s: %s", room, username, message);
        
        for (Session s : session.getOpenSessions()) {
            // 广播给同房间用户
            try {
                s.getBasicRemote().sendText(formatted);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

编码器和解码器​

Jakarta WebSocket 提供了编码器和解码器机制，用于在 Java 对象和 WebSocket 消息之间转换。

编码器（Encoder）​

编码器将 Java 对象转换为 WebSocket 消息格式：

import jakarta.websocket.Encoder;
import jakarta.websocket.EndpointConfig;
import com.fasterxml.jackson.databind.ObjectMapper;

// 文本编码器
public class MessageEncoder implements Encoder.Text<Message> {
    
    private ObjectMapper mapper;
    
    @Override
    public void init(EndpointConfig config) {
        mapper = new ObjectMapper();
    }
    
    @Override
    public void destroy() {
        // 清理资源
    }
    
    @Override
    public String encode(Message message) throws EncodeException {
        try {
            return mapper.writeValueAsString(message);
        } catch (Exception e) {
            throw new EncodeException(message, "编码失败", e);
        }
    }
}

// 二进制编码器
public class MessageBinaryEncoder implements Encoder.Binary<Message> {
    
    @Override
    public ByteBuffer encode(Message message) throws EncodeException {
        // 将对象编码为 ByteBuffer
        String json = new ObjectMapper().writeValueAsString(message);
        return ByteBuffer.wrap(json.getBytes(StandardCharsets.UTF_8));
    }
    
    @Override
    public void init(EndpointConfig config) {}
    
    @Override
    public void destroy() {}
}

解码器（Decoder）​

解码器将 WebSocket 消息转换为 Java 对象：

import jakarta.websocket.Decoder;
import jakarta.websocket.EndpointConfig;
import com.fasterxml.jackson.databind.ObjectMapper;

public class MessageDecoder implements Decoder.Text<Message> {
    
    private ObjectMapper mapper;
    
    @Override
    public void init(EndpointConfig config) {
        mapper = new ObjectMapper();
    }
    
    @Override
    public void destroy() {
        // 清理资源
    }
    
    @Override
    public Message decode(String s) throws DecodeException {
        try {
            return mapper.readValue(s, Message.class);
        } catch (Exception e) {
            throw new DecodeException(s, "解码失败", e);
        }
    }
    
    @Override
    public boolean willDecode(String s) {
        // 检查是否可以解码
        try {
            mapper.readTree(s);
            return true;
        } catch (Exception e) {
            return false;
        }
    }
}

使用编解码器​

在 @ServerEndpoint 注解中指定编解码器：

@ServerEndpoint(
    value = "/message",
    encoders = {MessageEncoder.class},
    decoders = {MessageDecoder.class}
)
public class MessageEndpoint {
    
    @OnMessage
    public void onMessage(Session session, Message message) {
        // message 已经是解码后的对象
        System.out.println("收到消息: " + message.getContent());
        
        try {
            // 发送对象（自动编码）
            Message response = new Message("server", "收到: " + message.getContent());
            session.getBasicRemote().sendObject(response);
        } catch (IOException | EncodeException e) {
            e.printStackTrace();
        }
    }
}

消息类定义​

public class Message {
    private String from;
    private String content;
    private long timestamp;
    
    public Message() {
        this.timestamp = System.currentTimeMillis();
    }
    
    public Message(String from, String content) {
        this();
        this.from = from;
        this.content = content;
    }
    
    // Getter 和 Setter
    public String getFrom() { return from; }
    public void setFrom(String from) { this.from = from; }
    
    public String getContent() { return content; }
    public void setContent(String content) { this.content = content; }
    
    public long getTimestamp() { return timestamp; }
    public void setTimestamp(long timestamp) { this.timestamp = timestamp; }
}

自定义配置器​

通过继承 ServerEndpointConfig.Configurator 类可以自定义端点配置。

访问 HTTP 请求信息​

import jakarta.websocket.server.ServerEndpointConfig;
import jakarta.websocket.server.HandshakeRequest;
import jakarta.websocket.HandshakeResponse;
import java.util.List;
import java.util.Map;

public class CustomConfigurator extends ServerEndpointConfig.Configurator {
    
    @Override
    public void modifyHandshake(ServerEndpointConfig sec,
                                HandshakeRequest request,
                                HandshakeResponse response) {
        // 获取 HTTP 请求头
        Map<String, List<String>> headers = request.getHeaders();
        
        // 获取查询参数
        String queryString = request.getRequestURI().getQuery();
        
        // 获取 Cookie
        Map<String, List<String>> cookies = request.getHeaders().get("Cookie");
        
        // 将请求信息存储到用户属性中
        sec.getUserProperties().put("httpRequest", request);
        sec.getUserProperties().put("headers", headers);
        
        // 修改响应头
        response.getHeaders().put("X-Custom-Header", 
            List.of("Custom Value"));
    }
}

Origin 验证​

public class OriginConfigurator extends ServerEndpointConfig.Configurator {
    
    private static final List<String> ALLOWED_ORIGINS = List.of(
        "http://localhost:3000",
        "https://example.com"
    );
    
    @Override
    public boolean checkOrigin(String originHeaderValue) {
        if (originHeaderValue == null) {
            return false;
        }
        return ALLOWED_ORIGINS.contains(originHeaderValue);
    }
}

使用配置器​

@ServerEndpoint(
    value = "/secure",
    configurator = CustomConfigurator.class
)
public class SecureEndpoint {
    
    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        // 获取存储的请求信息
        HandshakeRequest request = (HandshakeRequest) 
            config.getUserProperties().get("httpRequest");
        
        Map<String, List<String>> headers = request.getHeaders();
        System.out.println("Origin: " + headers.get("Origin"));
    }
}

Spring Boot 集成​

Spring Boot 提供了更简洁的 WebSocket 支持，可以与 Spring 生态系统无缝集成。

添加依赖​

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

WebSocket 配置​

import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.*;

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new ChatHandler(), "/ws/chat")
            .setAllowedOrigins("*");
    }
}

处理器实现​

import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArrayList;

public class ChatHandler extends TextWebSocketHandler {
    
    private final List<WebSocketSession> sessions = new CopyOnWriteArrayList<>();
    
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        sessions.add(session);
        System.out.println("连接建立: " + session.getId());
    }
    
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        System.out.println("收到消息: " + payload);
        
        // 广播给所有连接
        for (WebSocketSession s : sessions) {
            if (s.isOpen()) {
                s.sendMessage(new TextMessage(payload));
            }
        }
    }
    
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        sessions.remove(session);
        System.out.println("连接关闭: " + session.getId());
    }
    
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        System.err.println("传输错误: " + exception.getMessage());
        if (session.isOpen()) {
            session.close();
        }
    }
}

使用 STOMP 协议​

Spring 支持 STOMP 子协议，提供更丰富的消息模式：

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.*;

@Configuration
@EnableWebSocketMessageBroker
public class StompWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        // 启用简单的内存消息代理
        config.enableSimpleBroker("/topic");
        
        // 客户端发送消息的前缀
        config.setApplicationDestinationPrefixes("/app");
    }
    
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
            .setAllowedOriginPatterns("*")
            .withSockJS();
    }
}

消息处理控制器​

import org.springframework.messaging.handler.annotation.*;
import org.springframework.stereotype.Controller;

@Controller
public class ChatController {
    
    @MessageMapping("/chat")
    @SendTo("/topic/messages")
    public ChatMessage send(ChatMessage message) {
        return new ChatMessage(message.getFrom(), message.getContent());
    }
    
    @MessageMapping("/chat/{room}")
    @SendTo("/topic/room/{room}")
    public ChatMessage sendToRoom(@DestinationVariable String room, 
                                   ChatMessage message) {
        return new ChatMessage(message.getFrom(), 
            "[" + room + "] " + message.getContent());
    }
}

消息类​

public class ChatMessage {
    private String from;
    private String content;
    
    public ChatMessage() {}
    
    public ChatMessage(String from, String content) {
        this.from = from;
        this.content = content;
    }
    
    public String getFrom() { return from; }
    public void setFrom(String from) { this.from = from; }
    
    public String getContent() { return content; }
    public void setContent(String content) { this.content = content; }
}

前端连接 STOMP​

import SockJS from 'sockjs-client';
import { Stomp } from '@stomp/stompjs';

// 连接到 STOMP 端点
const socket = new SockJS('/ws');
const stompClient = Stomp.over(socket);

stompClient.connect({}, (frame) => {
    console.log('已连接:', frame);
    
    // 订阅主题
    stompClient.subscribe('/topic/messages', (message) => {
        const body = JSON.parse(message.body);
        console.log(`${body.from}: ${body.content}`);
    });
    
    // 发送消息
    stompClient.send('/app/chat', {}, JSON.stringify({
        from: '张三',
        content: '大家好'
    }));
});

完整聊天室示例​

服务端代码​

import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

@ServerEndpoint("/chatroom/{username}")
public class ChatRoomEndpoint {
    
    // 存储所有房间和会话
    private static final Map<String, Set<Session>> rooms = new ConcurrentHashMap<>();
    
    @OnOpen
    public void onOpen(Session session, @PathParam("username") String username) {
        // 从查询参数获取房间名
        String room = getQueryParam(session, "room", "general");
        
        // 存储用户信息
        session.getUserProperties().put("username", username);
        session.getUserProperties().put("room", room);
        
        // 添加到房间
        rooms.computeIfAbsent(room, k -> ConcurrentHashMap.newKeySet())
             .add(session);
        
        // 广播加入消息
        broadcast(room, String.format("系统: %s 加入了聊天室", username));
        
        // 发送欢迎消息
        sendTo(session, "系统: 欢迎 " + username + "，当前在线 " + 
            rooms.get(room).size() + " 人");
    }
    
    @OnMessage
    public void onMessage(Session session, String message) {
        String username = (String) session.getUserProperties().get("username");
        String room = (String) session.getUserProperties().get("room");
        
        // 广播消息
        broadcast(room, String.format("%s: %s", username, message));
    }
    
    @OnClose
    public void onClose(Session session) {
        String username = (String) session.getUserProperties().get("username");
        String room = (String) session.getUserProperties().get("room");
        
        // 从房间移除
        Set<Session> roomSessions = rooms.get(room);
        if (roomSessions != null) {
            roomSessions.remove(session);
            if (roomSessions.isEmpty()) {
                rooms.remove(room);
            }
        }
        
        // 广播离开消息
        broadcast(room, String.format("系统: %s 离开了聊天室", username));
    }
    
    @OnError
    public void onError(Session session, Throwable error) {
        System.err.println("WebSocket 错误: " + error.getMessage());
    }
    
    private void broadcast(String room, String message) {
        Set<Session> roomSessions = rooms.get(room);
        if (roomSessions == null) return;
        
        for (Session session : roomSessions) {
            sendTo(session, message);
        }
    }
    
    private void sendTo(Session session, String message) {
        if (session.isOpen()) {
            try {
                session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    private String getQueryParam(Session session, String name, String defaultValue) {
        String query = session.getQueryString();
        if (query == null) return defaultValue;
        
        for (String param : query.split("&")) {
            String[] pair = param.split("=");
            if (pair.length == 2 && pair[0].equals(name)) {
                return pair[1];
            }
        }
        return defaultValue;
    }
}

前端代码​

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>聊天室</title>
    <style>
        body {
            font-family: Arial, sans-serif;
            max-width: 800px;
            margin: 0 auto;
            padding: 20px;
        }
        #messages {
            height: 400px;
            overflow-y: auto;
            border: 1px solid #ccc;
            padding: 10px;
            margin-bottom: 10px;
        }
        .message {
            margin: 5px 0;
            padding: 5px;
        }
        .system {
            color: #666;
            font-style: italic;
        }
        .mine {
            background-color: #e8f5e9;
            margin-left: 20%;
        }
        .others {
            background-color: #f5f5f5;
            margin-right: 20%;
        }
        #input {
            width: 70%;
            padding: 10px;
        }
        button {
            padding: 10px 20px;
        }
    </style>
</head>
<body>
    <h1>聊天室</h1>
    <div id="messages"></div>
    <input type="text" id="input" placeholder="输入消息...">
    <button onclick="sendMessage()">发送</button>

    <script>
        const username = prompt('请输入用户名:') || '匿名';
        const room = new URLSearchParams(window.location.search).get('room') || 'general';
        
        const ws = new WebSocket(`ws://${location.host}/chatroom/${username}?room=${room}`);
        const messagesDiv = document.getElementById('messages');
        const input = document.getElementById('input');
        
        ws.onopen = () => {
            console.log('已连接');
        };
        
        ws.onmessage = (event) => {
            const message = event.data;
            const div = document.createElement('div');
            div.className = 'message';
            
            if (message.startsWith('系统:')) {
                div.classList.add('system');
                div.textContent = message;
            } else {
                const colonIndex = message.indexOf(':');
                const sender = message.substring(0, colonIndex);
                const content = message.substring(colonIndex + 2);
                
                div.classList.add(sender === username ? 'mine' : 'others');
                div.innerHTML = `<strong>${sender}</strong>: ${content}`;
            }
            
            messagesDiv.appendChild(div);
            messagesDiv.scrollTop = messagesDiv.scrollHeight;
        };
        
        ws.onclose = () => {
            console.log('连接关闭');
        };
        
        ws.onerror = (error) => {
            console.error('连接错误:', error);
        };
        
        function sendMessage() {
            const content = input.value.trim();
            if (content && ws.readyState === WebSocket.OPEN) {
                ws.send(content);
                input.value = '';
            }
        }
        
        input.addEventListener('keypress', (e) => {
            if (e.key === 'Enter') {
                sendMessage();
            }
        });
    </script>
</body>
</html>

小结​

本章介绍了 Java WebSocket 服务端的多种实现方式：

  1. Jakarta WebSocket API：Java 标准 WebSocket API

    • 注解式端点：使用 @ServerEndpoint 等注解，简单易用
    • 编程式端点：继承 Endpoint 类，提供更多控制

  2. 核心概念：

    • Session 管理：连接状态、用户数据存储
    • 路径参数：从 URL 中提取动态值
    • 编码器和解码器：对象与消息的转换

  3. Spring Boot 集成：

    • 基本 WebSocket 处理器
    • STOMP 协议支持

选择建议：

  • 标准 Java EE/Jakarta EE 项目：使用 Jakarta WebSocket API
  • Spring Boot 项目：使用 Spring WebSocket 支持
  • 需要复杂消息模式：使用 STOMP 子协议