SpringBoot实现WebSocket即时通讯的示例代码

 org.springframework.bootspring-boot-starter-websocket org.projectlomboklombok com.alibabafastjson1.2.3

package com.shucha.deveiface.web.config; /** * @author tqf * @Description * @Version 1.0 * @since 2022-04-12 15:35 */ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.server.standard.ServerEndpointExporter; /** * 开启WebSocket */ @Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter serverEndpointExporter(){ return new ServerEndpointExporter(); } } 

package com.shucha.deveiface.web.ws; /** * @author tqf * @Description * @Version 1.0 * @since 2022-04-12 15:33 */ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.springframework.web.socket.WebSocketSession; import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArraySet; @Component @ServerEndpoint("/webSocket/{userId}") @Slf4j public class WebSocketServer { private Session session; private String userId; /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/ private static int onlineCount = 0; private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet<>(); /** * concurrent包的线程安全set，用来存放每个客户端对应的MyWebSocket对象 */ private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap(); /** * 为了保存在线用户信息，在方法中新建一个list存储一下【实际项目依据复杂度，可以存储到数据库或者缓存】 */ private final static List SESSIONS = Collections.synchronizedList(new ArrayList<>()); /** * 建立连接 * @param session * @param userId */ @OnOpen public void onOpen(Session session, @PathParam("userId") String userId) { this.session = session; this.userId = userId; webSocketSet.add(this); SESSIONS.add(session); if (webSocketMap.containsKey(userId)) { webSocketMap.remove(userId); webSocketMap.put(userId,this); } else { webSocketMap.put(userId,this); addOnlineCount(); } // log.info("【websocket消息】有新的连接, 总数:{}", webSocketSet.size()); log.info("[连接ID:{}] 建立连接, 当前连接数:{}", this.userId, webSocketMap.size()); } /** * 断开连接 */ @OnClose public void onClose() { webSocketSet.remove(this); if (webSocketMap.containsKey(userId)) { webSocketMap.remove(userId); subOnlineCount(); } // log.info("【websocket消息】连接断开, 总数:{}", webSocketSet.size()); log.info("[连接ID:{}] 断开连接, 当前连接数:{}", userId, webSocketMap.size()); } /** * 发送错误 * @param session * @param error */ @OnError public void onError(Session session, Throwable error) { log.info("[连接ID:{}] 错误原因:{}", this.userId, error.getMessage()); error.printStackTrace(); } /** * 收到消息 * @param message */ @OnMessage public void onMessage(String message) { // log.info("【websocket消息】收到客户端发来的消息:{}", message); log.info("[连接ID:{}] 收到消息:{}", this.userId, message); } /** * 发送消息 * @param message * @param userId */ public void sendMessage(String message,Long userId) { WebSocketServer webSocketServer = webSocketMap.get(String.valueOf(userId)); if (webSocketServer!=null){ log.info("【websocket消息】推送消息, message={}", message); try { webSocketServer.session.getBasicRemote().sendText(message); } catch (Exception e) { e.printStackTrace(); log.error("[连接ID:{}] 发送消息失败, 消息:{}", this.userId, message, e); } } } /** * 群发消息 * @param message */ public void sendMassMessage(String message) { try { for (Session session : SESSIONS) { if (session.isOpen()) { session.getBasicRemote().sendText(message); log.info("[连接ID:{}] 发送消息:{}",session.getRequestParameterMap().get("userId"),message); } } } catch (Exception e) { e.printStackTrace(); } } /** * 获取当前连接数 * @return */ public static synchronized int getOnlineCount() { return onlineCount; } /** * 当前连接数加一 */ public static synchronized void addOnlineCount() { WebSocketServer.onlineCount++; } /** * 当前连接数减一 */ public static synchronized void subOnlineCount() { WebSocketServer.onlineCount--; } }

package com.shucha.deveiface.web.controller; import com.alibaba.fastjson.JSONObject; import com.shucha.deveiface.web.ws.WebSocketServer; import lombok.Data; import lombok.experimental.Accessors; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; /** * @author tqf * @Description * @Version 1.0 * @since 2022-04-12 15:44 */ @RestController @RequestMapping("/web") public class TestWebSocket { @Autowired private WebSocketServer webSocketServer; /** * 消息发送测试 */ @GetMapping("/test") public void test(){ for (int i=1;i<4;i++) { WebsocketResponse response = new WebsocketResponse(); response.setUserId(String.valueOf(i)); response.setUserName("姓名"+ i); response.setAge(i); webSocketServer.sendMessage(JSONObject.toJSONString(response), Long.valueOf(String.valueOf(i))); } } /** * 群发消息测试(给当前连接用户发送) */ @GetMapping("/sendMassMessage") public void sendMassMessage(){ WebsocketResponse response = new WebsocketResponse(); response.setUserName("群发消息模板测试"); webSocketServer.sendMassMessage(JSONObject.toJSONString(response)); } @Data @Accessors(chain = true) public static class WebsocketResponse { private String userId; private String userName; private int age; } }