SpringBoot中使用Netty开发WebSocket服务改造多线程群发消息示例代码

package com.badao.demo.websocket; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.timeout.IdleStateEvent; import lombok.extern.slf4j.Slf4j; import org.yeauty.annotation.*; import org.yeauty.pojo.Session; import java.io.IOException; import java.util.concurrent.Semaphore; @ServerEndpoint(path = "/websocket/{userName}", port = "${ws.port}", readerIdleTimeSeconds = "${ws.readerIdleTimeSeconds}", writerIdleTimeSeconds = "${ws.writerIdleTimeSeconds}", allIdleTimeSeconds = "${ws.allIdleTimeSeconds}") @Slf4j public class WebSocketServer { /** * 默认最多允许同时在线人数 200 */ public static int socketMaxOnlineCount = 200; private static final Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount); @OnOpen public void onOpen(Session session, HttpHeaders headers, @PathVariable String userName) { boolean semaphoreFlag = false; // 尝试获取信号量 semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore); if (!semaphoreFlag) { // 未获取到信号量 log.error("\n 当前人数 - {} , 限制人数：{} ", WebSocketUsers.getUsers().size(), socketMaxOnlineCount); WebSocketUsers.sendMessageToUserByText(session, "当前人数：" + WebSocketUsers.getUsers().size() + " 限制人数：" + socketMaxOnlineCount); session.close(); } else { // 添加用户 WebSocketUsers.sendMessageToUserByText(session, "连接成功"); WebSocketUsers.put(userName, session); log.warn("\n 用户：{} 连接后 , 当前人数 : {}", userName, WebSocketUsers.getUsers().size()); } } @OnClose public void onClose(Session session, @PathVariable String userName) throws IOException { // 移除用户 WebSocketUsers.remove(userName); log.warn("\n 用户：{} 关闭后 , 当前人数 : {}", userName, WebSocketUsers.getUsers().size()); // 获取到信号量则需释放 SemaphoreUtils.release(socketSemaphore); } @OnError public void onError(Session session, @PathVariable String userName, Throwable exception) { if (session.isOpen()) { // 关闭连接 session.close(); } log.warn("\n 连接异常 - {}", userName); log.warn("\n 异常信息 - {}", exception); // 移出用户 WebSocketUsers.remove(userName); // 获取到信号量则需释放 SemaphoreUtils.release(socketSemaphore); } @OnMessage public void onMessage(Session session, String message) { WebSocketUsers.sendMessageToUserByText(session, message); } @OnBinary public void onBinary(Session session, byte[] bytes) { for (byte b : bytes) { System.out.println(b); } session.sendBinary(bytes); } @OnEvent public void onEvent(@PathVariable String userName, Session session, Object evt) { if (evt instanceof IdleStateEvent) { IdleStateEvent idleStateEvent = (IdleStateEvent) evt; switch (idleStateEvent.state()) { case READER_IDLE: log.error("user-{} Read timeout！", userName); session.close(); break; case WRITER_IDLE: log.error("user-{} Write timeout！", userName); session.close(); break; case ALL_IDLE: log.error("user-{} All timeout！", userName); session.close(); break; default: break; } } } }