Java实现WebSocket案例

wen java案例 2026-08-05 2

本文目录导读：

Java实现WebSocket案例

项目结构
Maven依赖 (pom.xml)
WebSocket服务端实现
消息模型和编码解码器
WebSocket客户端实现
服务器启动类
配置文件（可选）
前端HTML测试页面
测试用例
运行说明
功能特性
注意事项

我来为您提供一个完整的Java WebSocket实现案例，使用标准的Java WebSocket API（JSR 356）和嵌入式服务器。

项目结构

websocket-demo/
├── pom.xml
└── src/
    └── main/java/com/example/websocket/
        ├── WebSocketServer.java
        ├── WebSocketClient.java
        ├── ChatMessage.java
        └── WebSocketServerApp.java

Maven依赖 (pom.xml)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>websocket-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <!-- Java EE WebSocket API -->
        <dependency>
            <groupId>javax.websocket</groupId>
            <artifactId>javax.websocket-api</artifactId>
            <version>1.1</version>
            <scope>provided</scope>
        </dependency>
        <!-- JSON 处理库 - 使用 Jackson -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.13.3</version>
        </dependency>
        <!-- Tyrus（参考实现）作为WebSocket客户端实现 -->
        <dependency>
            <groupId>org.glassfish.tyrus</groupId>
            <artifactId>tyrus-client</artifactId>
            <version>1.15</version>
            <scope>test</scope>
        </dependency>
        <!-- Log4j2 -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.17.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>2.17.1</version>
        </dependency>
        <!-- 测试依赖 -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>11</source>
                    <target>11</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

WebSocket服务端实现

package com.example.websocket;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Logger;
@ServerEndpoint(value = "/chat/{username}", 
                encoders = {MessageEncoder.class}, 
                decoders = {MessageDecoder.class},
                configurator = CustomConfigurator.class)
public class WebSocketServer {
    private static final Logger logger = Logger.getLogger(WebSocketServer.class.getName());
    // 存储所有在线客户端会话
    private static final Map<String, Session> onlineUsers = new ConcurrentHashMap<>();
    private String username;
    /**
     * 连接建立时调用
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("username") String username) {
        this.username = username;
        onlineUsers.put(username, session);
        logger.info("用户 " + username + " 已连接");
        // 广播上线消息
        ChatMessage message = new ChatMessage("SYSTEM", "User " + username + " joined the chat!");
        broadcastToAll(message);
        // 发送在线用户列表给新连接的用户
        sendOnlineUsers(session);
    }
    /**
     * 接收客户端消息时调用
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        logger.info("从 " + username + " 收到消息: " + message);
        try {
            // 如果是JSON字符串，转换为ChatMessage对象
            ChatMessage chatMessage = MessageUtil.fromJson(message);
            chatMessage.setSender(username);
            chatMessage.setTimestamp(System.currentTimeMillis());
            // 广播消息
            broadcastToAll(chatMessage);
        } catch (Exception e) {
            logger.warning("消息处理失败: " + e.getMessage());
            sendErrorMessage(session, "消息格式不正确");
        }
    }
    /**
     * 连接关闭时调用
     */
    @OnClose
    public void onClose(Session session, CloseReason reason) {
        onlineUsers.remove(username);
        logger.info("用户 " + username + " 断开连接，原因: " + reason.getReasonPhrase());
        // 广播离线消息
        ChatMessage message = new ChatMessage("SYSTEM", "User " + username + " has left the chat.");
        broadcastToAll(message);
    }
    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(Session session, Throwable error) {
        logger.severe("WebSocket错误: " + error.getMessage());
        error.printStackTrace();
        try {
            if (session.isOpen()) {
                session.close(new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, error.getMessage()));
            }
        } catch (IOException e) {
            logger.severe("关闭连接失败: " + e.getMessage());
        }
    }
    /**
     * 广播消息给所有在线用户
     */
    private void broadcastToAll(ChatMessage message) {
        String jsonMessage = MessageUtil.toJson(message);
        onlineUsers.values().forEach(session -> {
            if (session.isOpen()) {
                try {
                    synchronized (session) {
                        session.getBasicRemote().sendText(jsonMessage);
                    }
                } catch (IOException e) {
                    logger.warning("发送消息失败给: " + session.getId() + ", 错误: " + e.getMessage());
                }
            }
        });
    }
    /**
     * 发送在线用户列表
     */
    private void sendOnlineUsers(Session session) {
        ChatMessage message = new ChatMessage("SYSTEM", "ONLINE_USERS:" + String.join(",", onlineUsers.keySet()));
        try {
            session.getBasicRemote().sendText(MessageUtil.toJson(message));
        } catch (IOException e) {
            logger.warning("发送在线列表失败: " + e.getMessage());
        }
    }
    /**
     * 发送错误消息给特定会话
     */
    private void sendErrorMessage(Session session, String error) {
        try {
            ChatMessage errorMessage = new ChatMessage("SYSTEM", "ERROR:" + error);
            session.getBasicRemote().sendText(MessageUtil.toJson(errorMessage));
        } catch (IOException e) {
            logger.severe("发送错误消息失败: " + e.getMessage());
        }
    }
    /**
     * 自定义配置器
     */
    public static class CustomConfigurator extends ServerEndpointConfig.Configurator {
        @Override
        public void modifyHandshake(ServerEndpointConfig config, 
                                   HandshakeRequest request, 
                                   HandshakeResponse response) {
            // 可以在这里添加额外的握手验证或配置
            logger.info("开始握手: " + request.getRequestURI());
        }
    }
}

消息模型和编码解码器

package com.example.websocket;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
public class ChatMessage {
    private String sender;
    private String content;
    private long timestamp;
    private String type;
    public ChatMessage() {}
    public ChatMessage(String sender, String content) {
        this.sender = sender;
        this.content = content;
        this.timestamp = System.currentTimeMillis();
        this.type = "MESSAGE";
    }
    // Getter和Setter方法
    public String getSender() { return sender; }
    public void setSender(String sender) { this.sender = sender; }
    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; }
    public String getType() { return type; }
    public void setType(String type) { this.type = type; }
    @Override
    public String toString() {
        return "ChatMessage{" +
                "sender='" + sender + '\'' +
                ", content='" + content + '\'' +
                ", timestamp=" + timestamp +
                ", type='" + type + '\'' +
                '}';
    }
}
/**
 * 消息编码器
 */
import javax.websocket.EncodeException;
import javax.websocket.Encoder;
import javax.websocket.EndpointConfig;
public class MessageEncoder implements Encoder.Text<ChatMessage> {
    @Override
    public String encode(ChatMessage message) throws EncodeException {
        return MessageUtil.toJson(message);
    }
    @Override
    public void init(EndpointConfig endpointConfig) {}
    @Override
    public void destroy() {}
}
/**
 * 消息解码器
 */
import javax.websocket.DecodeException;
import javax.websocket.Decoder;
import javax.websocket.EndpointConfig;
public class MessageDecoder implements Decoder.Text<ChatMessage> {
    @Override
    public ChatMessage decode(String s) throws DecodeException {
        return MessageUtil.fromJson(s);
    }
    @Override
    public boolean willDecode(String s) {
        return s != null;
    }
    @Override
    public void init(EndpointConfig endpointConfig) {}
    @Override
    public void destroy() {}
}
/**
 * JSON消息工具类
 */
public class MessageUtil {
    private static final ObjectMapper objectMapper = new ObjectMapper();
    public static String toJson(ChatMessage message) {
        try {
            return objectMapper.writeValueAsString(message);
        } catch (IOException e) {
            e.printStackTrace();
            return "{}";
        }
    }
    public static ChatMessage fromJson(String json) {
        try {
            return objectMapper.readValue(json, ChatMessage.class);
        } catch (IOException e) {
            e.printStackTrace();
            return new ChatMessage("UNKNOWN", json);
        }
    }
}

WebSocket客户端实现

package com.example.websocket;
import javax.websocket.*;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Scanner;
import java.util.logging.Logger;
@ClientEndpoint
public class WebSocketClient {
    private static final Logger logger = Logger.getLogger(WebSocketClient.class.getName());
    private Session session;
    private String username;
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        logger.info("连接到服务器");
        System.out.println("已连接到服务器");
    }
    @OnMessage
    public void onMessage(String message) {
        System.out.println("收到消息: " + message);
    }
    @OnClose
    public void onClose(Session session, CloseReason reason) {
        logger.info("连接关闭: " + reason.getReasonPhrase());
        System.out.println("连接已关闭");
    }
    @OnError
    public void onError(Session session, Throwable error) {
        logger.severe("客户端错误: " + error.getMessage());
    }
    /**
     * 连接到WebSocket服务器
     */
    public void connect(String url) throws URISyntaxException, IOException, DeploymentException {
        WebSocketContainer container = ContainerProvider.getWebSocketContainer();
        session = container.connectToServer(this, new URI(url));
    }
    /**
     * 发送消息
     */
    public void sendMessage(String message) throws IOException {
        if (session != null && session.isOpen()) {
            session.getBasicRemote().sendText(message);
        } else {
            throw new IOException("WebSocket会话未打开");
        }
    }
    /**
     * 断开连接
     */
    public void close() throws IOException {
        if (session != null) {
            session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "客户端关闭"));
        }
    }
    /**
     * 客户端启动器 - 命令行交互模式
     */
    public static void main(String[] args) {
        Scanner scanner = new Scanner(System.in);
        System.out.print("请输入用户名: ");
        String username = scanner.nextLine();
        WebSocketClient client = new WebSocketClient();
        try {
            // 连接到服务器（默认端口8080）
            String serverUrl = "ws://localhost:8080/websocket/chat/" + username;
            client.connect(serverUrl);
            System.out.println("输入消息进行聊天，输入 'quit' 退出");
            while (true) {
                System.out.print("> ");
                String input = scanner.nextLine();
                if (input.equalsIgnoreCase("quit")) {
                    client.close();
                    break;
                }
                client.sendMessage(input);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            scanner.close();
        }
    }
}

服务器启动类

package com.example.websocket;
import org.apache.catalina.Context;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.startup.Tomcat;
import javax.servlet.ServletException;
import javax.websocket.server.ServerContainer;
import javax.websocket.server.ServerEndpointConfig;
import java.io.File;
import java.util.logging.Logger;
/**
 * 嵌入式Tomcat服务器启动类
 */
public class WebSocketServerApp {
    private static final Logger logger = Logger.getLogger(WebSocketServerApp.class.getName());
    public static void main(String[] args) throws Exception {
        int port = 8080;
        String contextPath = "/websocket";
        // 创建Tomcat实例
        Tomcat tomcat = new Tomcat();
        tomcat.setPort(port);
        tomcat.setBaseDir(System.getProperty("java.io.tmpdir"));
        tomcat.getConnector();
        // 设置Host配置
        tomcat.getHost().setAppBase(".");
        tomcat.getHost().setAutoDeploy(true);
        // 创建Context
        Context context = tomcat.addContext(contextPath, new File(".").getAbsolutePath());
        // 添加自定义Servlet（可选，用于HTTP请求处理）
        Tomcat.addServlet(context, "helloServlet", new HelloServlet());
        context.addServletMappingDecoded("/hello", "helloServlet");
        // 添加WebSocket支持
        ServerContainer serverContainer = (ServerContainer) tomcat.getServletContext()
                .getAttribute(ServerContainer.class.getName());
        if (serverContainer != null) {
            // 配置WebSocket连接超时和缓冲区大小
            serverContainer.setDefaultMaxSessionIdleTimeout(3000); // 3秒
            serverContainer.setDefaultMaxSessionTimeout(300000); // 5分钟超时
            // 注册WebSocket端点
            serverContainer.addEndpoint(WebSocketServer.class);
            logger.info("WebSocket端点已注册");
        }
        // 启动Tomcat
        tomcat.start();
        logger.info("服务器启动成功，监听端口: " + port + ", 路径: " + contextPath);
        logger.info("WebSocket连接地址: ws://localhost:" + port + contextPath + "/chat/{username}");
        // 等待关闭
        tomcat.getServer().await();
    }
    /**
     * 简单的HTTP Servlet
     */
    public static class HelloServlet extends javax.servlet.http.HttpServlet {
        @Override
        protected void doGet(javax.servlet.http.HttpServletRequest req, 
                           javax.servlet.http.HttpServletResponse resp) 
                throws javax.servlet.ServletException, java.io.IOException {
            resp.setContentType("text/html;charset=UTF-8");
            resp.getWriter().println("<h1>你好，WebSocket测试页面</h1>");
            resp.getWriter().println("<h2>WebSocket端点: /websocket/chat/{username}</h2>");
        }
    }
}

配置文件（可选）

创建 src/main/resources/log4j2.xml:

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
    <Properties>
        <Property name="logPath">logs</Property>
        <Property name="consolePatternLayout">%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n</Property>
    </Properties>
    <Appenders>
        <Console name="Console" target="SYSTEM_OUT">
            <PatternLayout pattern="${consolePatternLayout}"/>
        </Console>
        <RollingFile name="RollingFile" fileName="${logPath}/websocket.log"
                     filePattern="${logPath}/websocket-%d{yyyy-MM-dd}-%i.log">
            <PatternLayout>
                <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger{36} - %msg%n</pattern>
            </PatternLayout>
            <Policies>
                <SizeBasedTriggeringPolicy size="10MB"/>
            </Policies>
        </RollingFile>
    </Appenders>
    <Loggers>
        <Root level="info">
            <AppenderRef ref="Console"/>
            <AppenderRef ref="RollingFile"/>
        </Root>
    </Loggers>
</Configuration>

前端HTML测试页面

创建 web/index.html:

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">WebSocket聊天室</title>
    <style>
        body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
        #messages { border: 1px solid #ccc; height: 400px; overflow-y: scroll; padding: 10px; margin-bottom: 10px; }
        .message { margin: 5px 0; padding: 5px; border-radius: 4px; background: #f1f1f1; }
        .system { color: #999; font-style: italic; }
        .private { background: #ffffee; border: 1px solid #cccc99; }
        input, button { padding: 8px; margin: 5px; }
        #messageInput { width: 70%; }
    </style>
</head>
<body>
    <h1>WebSocket聊天室</h1>
    <div id="loginSection">
        <input type="text" id="username" placeholder="手机号" />
        <button onclick="connect()">连接</button>
    </div>
    <div id="chatSection" style="display: none;">
        <div id="messages"></div>
        <div>
            <input type="text" id="messageInput" placeholder="输入消息...">
            <input type="text" id="targetUser" placeholder="目标用户(留空为广播)">
            <button onclick="sendMessage()">发送</button>
            <button onclick="disconnect()">断开连接</button>
        </div>
    </div>
    <script>
        let ws = null;
        function connect() {
            const username = document.getElementById('username').value.trim();
            if (!username) {
                alert('请输入用户名');
                return;
            }
            const wsUrl = `ws://localhost:8080/websocket/chat/${username}`;
            // 创建WebSocket连接
            ws = new WebSocket(wsUrl);
            // 连接建立事件
            ws.onopen = function(event) {
                console.log('连接成功');
                document.getElementById('loginSection').style.display = 'none';
                document.getElementById('chatSection').style.display = 'block';
                addMessage('已连接到服务器');
            };
            // 接收消息事件
            ws.onmessage = function(event) {
                console.log('收到消息:', event.data);
                try {
                    const data = JSON.parse(event.data);
                    // 处理系统消息
                    if (data.sender === 'SYSTEM' && data.content.startsWith('ONLINE_USERS:')) {
                        const users = data.content.substring('ONLINE_USERS:'.length);
                        addMessage(`在线用户: ${users}`, 'system');
                    } else if (data.sender === 'SYSTEM' && data.content.startsWith('ERROR:')) {
                        addMessage(`错误: ${data.content.substring(6)}`, 'system');
                    } else {
                        // 普通聊天消息
                        addMessage(`${data.sender}: ${data.content}`);
                    }
                } catch (e) {
                    addMessage(event.data);
                }
            };
            // 连接关闭事件
            ws.onclose = function(event) {
                console.log('连接关闭');
                addMessage('连接已关闭', 'system');
                document.getElementById('loginSection').style.display = 'block';
                document.getElementById('chatSection').style.display = 'none';
            };
            // 错误事件
            ws.onerror = function(event) {
                console.error('WebSocket错误:', event);
                addMessage('连接错误: ' + event.message, 'system');
            };
        }
        function sendMessage() {
            if (!ws || ws.readyState === WebSocket.CLOSED) {
                alert('请先连接');
                return;
            }
            const content = document.getElementById('messageInput').value.trim();
            const targetUser = document.getElementById('targetUser').value.trim();
            if (!content) {
                alert('请输入消息');
                return;
            }
            const message = {
                sender: document.getElementById('username').value.trim(),
                content: targetUser ? `@${targetUser} ${content}` : content,
                timestamp: Date.now(),
                type: 'MESSAGE'
            };
            ws.send(JSON.stringify(message));
            document.getElementById('messageInput').value = '';
        }
        function disconnect() {
            if (ws) {
                ws.close();
            }
        }
        function addMessage(text, type = '') {
            const messagesDiv = document.getElementById('messages');
            const messageEl = document.createElement('div');
            messageEl.className = `message ${type}`;
            messageEl.textContent = text;
            messagesDiv.appendChild(messageEl);
            messagesDiv.scrollTop = messagesDiv.scrollHeight;
        }
        // 回车发送消息
        document.getElementById('messageInput').addEventListener('keypress', function(e) {
            if (e.key === 'Enter') {
                sendMessage();
            }
        });
    </script>
</body>
</html>

测试用例

package com.example.websocket.test;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import javax.websocket.*;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
import static org.junit.Assert.*;
public class WebSocketTest {
    private static final Logger logger = Logger.getLogger(WebSocketTest.class.getName());
    private static final String WS_URL = "ws://localhost:8080/websocket/chat/testUser";
    private CountDownLatch latch;
    private String receivedMessage;
    private Session session;
    @Before
    public void setup() throws Exception {
        // 这里假设服务器已经在运行
        // 在实际测试中，可以启动嵌入式服务器
    }
    @Test
    public void testWebSocketConnection() throws Exception {
        latch = new CountDownLatch(1);
        WebSocketContainer container = ContainerProvider.getWebSocketContainer();
        // 使用匿名客户端连接
        Session testSession = container.connectToServer(new Endpoint() {
            @Override
            public void onOpen(Session session, EndpointConfig config) {
                logger.info("测试连接已打开");
                // 发送测试消息
                try {
                    session.getBasicRemote().sendText("Hello WebSocket!");
                } catch (IOException e) {
                    fail("发送消息失败: " + e.getMessage());
                }
            }
            @Override
            public void onMessage(Session session, String message) {
                logger.info("收到消息: " + message);
                receivedMessage = message;
                latch.countDown();
            }
            @Override
            public void onError(Session session, Throwable error) {
                logger.severe("错误: " + error.getMessage());
                fail("WebSocket错误: " + error.getMessage());
            }
            @Override
            public void onClose(Session session, CloseReason closeReason) {
                logger.info("连接关闭: " + closeReason.getReasonPhrase());
            }
        }, new URI(WS_URL));
        // 等待消息接收
        assertTrue(latch.await(5, TimeUnit.SECONDS));
        assertNotNull(receivedMessage);
        assertTrue(receivedMessage.contains("Hello WebSocket!"));
    }
    @Test
    public void testMessageEncoding() {
        ChatMessage message = new ChatMessage("sender", "Hello");
        String json = MessageUtil.toJson(message);
        assertNotNull(json);
        assertTrue(json.contains("sender"));
        assertTrue(json.contains("Hello"));
    }
    @After
    public void cleanup() {
        if (session != null) {
            try {
                session.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

运行说明

1 启动服务器

mvn clean package
java -jar target/websocket-demo-1.0-SNAPSHOT.jar

2 测试连接

使用浏览器打开 web/index.html 测试页面

使用命令行客户端：

java -cp target/websocket-demo-1.0-SNAPSHOT.jar com.example.websocket.WebSocketClient

3 使用工具测试

可以使用以下工具进行测试：

Postman: 支持WebSocket测试
wscat: Node.js的WebSocket客户端工具

# 安装wscat
npm install -g wscat
# 连接测试
wscat -c ws://localhost:8080/websocket/chat/testUser

功能特性

✅ 完整的WebSocket实现
✅ 消息编解码
✅ 广播消息
✅ 在线用户管理
✅ 错误处理
✅ 前端测试页面
✅ 嵌入式服务器
✅ 单元测试

注意事项

安全性: 实际生产环境需要添加认证和授权
性能: 对于大规模用户，需要考虑集群支持和负载均衡
心跳机制: 建议添加心跳检测来处理断开的连接
消息大小: 注意配置WebSocket消息大小限制

这个案例提供了完整的WebSocket解决方案，包含服务器、客户端、前端页面和测试用例,可以直接运行和学习。