Springboot整合Websocket实现ws和wss连接
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Component
@Slf4j
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {
/**线程安全Map*/
private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
//==========【websocket接受、推送消息等方法 —— 具体服务节点推送ws消息】=====================================================
@OnOpen
public void onOpen(Session session, @PathParam(value = "userId") String userId) {
try {
sessionPool.put(userId, session);
log.info("【系统 WebSocket】有新的连接，总数为:" + sessionPool.size());
} catch (Exception e) {
}
}
@OnClose
public void onClose(@PathParam("userId") String userId) {
try {
sessionPool.remove(userId);
log.info("【系统 WebSocket】连接断开，总数为:" + sessionPool.size());
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* ws推送消息
*
* @param userId
* @param message
*/
public void pushMessage(String userId, String message) {
for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
//userId key值= {用户id + "_"+ 登录token的md5串}
if (item.getKey().contains(userId)) {
Session session = item.getValue();
try {
synchronized (session){
log.info("【系统 WebSocket】推送单人消息:" + message);
session.getBasicRemote().sendText(message);
}
} catch (Exception e) {
log.error(e.getMessage(),e);
}
}
}
}
/**
* ws遍历群发消息
*/
public void pushMessage(String message) {
try {
for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
try {
item.getValue().getAsyncRemote().sendText(message);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
log.info("【系统 WebSocket】广播消息:" + message);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
/**
* ws接受客户端消息
*/
@OnMessage
public void onMessage(String message, @PathParam(value = "userId") String userId) {
if(!"ping".equals(message) && !WebsocketConst.CMD_CHECK.equals(message)){
log.info("【系统 WebSocket】收到客户端消息:" + message);
}else{
log.debug("【系统 WebSocket】收到客户端消息:" + message);
}
}
/**
* 配置错误信息处理
*
* @param session
* @param t
*/
@OnError
public void onError(Session session, Throwable t) {
log.warn("【系统 WebSocket】消息出现错误");
t.printStackTrace();
}
//==========【系统 WebSocket接受、推送消息等方法 —— 具体服务节点推送ws消息】=================================================
}