引言：WebSocket在現代應用中的重要性
在當今實時交互應用盛行的時代，WebSocket協議已成為實現雙向通信的核心技術。相比傳統的HTTP輪詢，WebSocket提供了：
- 真正的全雙工通信
- 極低的延遲（毫秒級）
- 高效的連接管理
- 減少不必要的網絡流量
本文將介紹如何使用netty-websocket-spring-boot-starter構建高性能WebSocket服務，實現消息收發功能。
一、Netty-WebSocket框架簡介
Netty作為高性能NIO框架，是構建WebSocket服務的理想選擇。netty-websocket-spring-boot-starter封裝了Netty的復雜配置，提供Spring Boot風格的開發體驗：
核心優勢：
- 高性能：基于Netty的Reactor模型，支持百萬級并發
- 簡化開發：注解驅動，類似Spring MVC
- 無縫集成：與Spring生態完美融合
- 可擴展性：支持自定義編解碼器和攔截器
<!-- Maven依賴 -->
<dependency><groupId>org.yeauty</groupId><artifactId>netty-websocket-spring-boot-starter</artifactId><version>0.11.0</version>
</dependency>
二、構建WebSocket服務端
1. 基礎服務端實現
@ServerEndpoint(path = "/chat", port = "8080")
@Component
public class ChatServer {private static final Map<String, Session> sessions = new ConcurrentHashMap<>();@OnOpenpublic void onOpen(Session session) {String clientId = session.id().asShortText();sessions.put(clientId, session);System.out.println("客戶端連接: " + clientId);}@OnClosepublic void onClose(Session session) {String clientId = session.id().asShortText();sessions.remove(clientId);System.out.println("客戶端斷開: " + clientId);}@OnMessagepublic void onMessage(Session session, String message) {System.out.println("收到消息: " + message);// 處理消息邏輯processMessage(session, message);}// 發送消息給指定客戶端public static void sendToClient(String clientId, String message) {Session session = sessions.get(clientId);if (session != null && session.isOpen()) {session.sendText(message);}}// 廣播消息public static void broadcast(String message) {sessions.values().forEach(session -> {if (session.isOpen()) {session.sendText(message);}});}
}
2. 核心注解解析
| 注解 | 說明 | 示例 |
|---|---|---|
@ServerEndpoint | 定義服務端點 | @ServerEndpoint(path="/ws", port="8080") |
@OnOpen | 連接建立時觸發 | public void onOpen(Session session) |
@OnClose | 連接關閉時觸發 | public void onClose(Session session) |
@OnMessage | 收到消息時觸發 | public void onMessage(String message) |
@OnError | 發生錯誤時觸發 | public void onError(Throwable error) |
三、消息收發實戰
1. 接收客戶端消息
@OnMessage
public void onMessage(Session session, String message) {try {// 解析JSON消息JsonNode json = new ObjectMapper().readTree(message);// 消息路由switch (json.get("type").asText()) {case "TEXT":handleTextMessage(session, json);break;case "IMAGE":handleImageMessage(session, json);break;case "COMMAND":handleCommand(session, json);break;default:sendError(session, "未知消息類型");}} catch (Exception e) {sendError(session, "消息格式錯誤");}
}private void handleTextMessage(Session session, JsonNode json) {String content = json.get("content").asText();String sender = json.get("sender").asText();// 業務處理邏輯MessageEntity message = messageService.saveMessage(sender, content);// 回復客戶端session.sendText("{\"status\":\"SUCCESS\",\"messageId\":" + message.getId() + "}");
}
2. 發送消息給客戶端
// 發送文本消息
public void sendTextMessage(String clientId, String content) {Session session = sessions.get(clientId);if (session != null && session.isOpen()) {JsonObject message = new JsonObject();message.addProperty("type", "TEXT");message.addProperty("content", content);message.addProperty("timestamp", System.currentTimeMillis());session.sendText(message.toString());}
}// 發送二進制數據（如圖片）
public void sendImage(String clientId, byte[] imageData) {Session session = sessions.get(clientId);if (session != null && session.isOpen()) {session.sendBinary(imageData);}
}// 帶回調的異步發送
public void sendWithCallback(String clientId, String message) {Session session = sessions.get(clientId);if (session != null && session.isOpen()) {session.sendText(message, new FutureCallback<Void>() {@Overridepublic void onSuccess(Void result) {log.info("消息發送成功");}@Overridepublic void onFailure(Throwable t) {log.error("消息發送失敗", t);// 重試邏輯}});}
}
四、高級功能實現
1. 心跳檢測機制
@OnEvent
public void onEvent(Session session, Object evt) {if (evt instanceof IdleStateEvent) {IdleStateEvent idleEvent = (IdleStateEvent) evt;if (idleEvent.state() == IdleState.READER_IDLE) {// 30秒無讀操作，發送心跳session.sendText("{\"type\":\"HEARTBEAT\"}");} else if (idleEvent.state() == IdleState.WRITER_IDLE) {// 60秒無寫操作，關閉連接session.close();}}
}
2. 消息壓縮傳輸
@OnMessage
public void onBinaryMessage(Session session, byte[] compressedData) {try {// 解壓縮消息String message = decompress(compressedData);// 處理消息...} catch (IOException e) {log.error("解壓縮失敗", e);}
}private String decompress(byte[] compressed) throws IOException {ByteArrayInputStream bis = new ByteArrayInputStream(compressed);GZIPInputStream gis = new GZIPInputStream(bis);return new String(gis.readAllBytes(), StandardCharsets.UTF_8);
}
3. 分布式會話管理
@Service
public class RedisSessionStore {@Autowiredprivate RedisTemplate<String, Object> redisTemplate;public void saveSession(String sessionId, SessionInfo info) {redisTemplate.opsForValue().set("ws:session:" + sessionId, info,1, TimeUnit.HOURS);}public SessionInfo getSessionInfo(String sessionId) {return (SessionInfo) redisTemplate.opsForValue().get("ws:session:" + sessionId);}
}// 會話信息類
@Data
public class SessionInfo {private String userId;private String deviceId;private String nodeId;private long lastActiveTime;
}
五、最佳實踐建議
-
連接管理優化
- 設置合理的最大連接數
- 實現連接數監控和告警
@Bean public ServerEndpointConfig config() {return ServerEndpointConfig.builder().port(8080).bossEventLoopGroup(2) // boss線程數.workerEventLoopGroup(16) // worker線程數.maxFramePayloadLength(1048576) // 1MB.build(); } -
安全防護措施
- 實現WSS(WebSocket Secure)
- 添加身份驗證
- 防止DDoS攻擊
@BeforeHandshake public void handshake(Session session, @RequestParam String token) {if (!authService.validate(token)) {session.close();} } -
性能監控指標
指標 說明 健康值 活動連接數 當前在線連接 < 80% 最大容量 消息吞吐量 消息/秒 根據業務調整 平均延遲 消息處理時間 < 100ms 錯誤率 失敗消息比例 < 0.1%
六、客戶端實現示例
// WebSocket客戶端
const socket = new WebSocket('wss://yourserver.com/chat');// 連接建立
socket.onopen = () => {console.log('連接已建立');// 發送文本消息socket.send(JSON.stringify({type: 'TEXT',content: '你好服務器!'}));
};// 接收消息
socket.onmessage = (event) => {const message = JSON.parse(event.data);console.log('收到消息:', message);
