外贸开发网站公司阿里指数网站
适用于单客户端，一个账号登陆一个客户端，登陆多个客户端会报错
The remote endpoint was in state [TEXT_FULL_WRITING]
这是因为此时的session是不同的，只能锁住一个session，解决此问题的方法把全局静态对象锁住，因为账号是唯一的
/*** @Description 开启springboot对websocket的支持* @Author WangKun* @Date 2023/8/14 17:21* @Version*/
@ConditionalOnProperty(name = "spring.profiles.active", havingValue = "dev")
@Configuration
public class WebSocketConfig{/*** @Description 注入一个ServerEndpointExporter, 会自动注册使用@ServerEndpoint注解* @param* @Throws* @Return org.springframework.web.socket.server.standard.ServerEndpointExporter* @Date 2023-08-14 17:26:31* @Author WangKun*/@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}
}
/*** @Description websocket服务，不考虑分组* @Author WangKun* @Date 2023/8/14 17:29* @Version*/
@ConditionalOnClass(value = WebSocketConfig.class)
@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocket {private static final long SESSION_TIMEOUT = 60000;//存放每个客户端对应的WebSocket对象。private static final ConcurrentHashMap<String, CopyOnWriteArraySet<WebSocket>> WEB_SOCKET_MAP = new ConcurrentHashMap<>();//与某个客户端的连接会话，需要通过它来给客户端发送数据private Session session;private String userId;/*** @param o* @Description 重写防止session重复* @Throws* @Return boolean* @Date 2023-09-01 10:02:51* @Author WangKun*/@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (o == null || getClass() != o.getClass()) {return false;}WebSocket that = (WebSocket) o;return Objects.equals(session, that.session);}@Overridepublic int hashCode() {return Objects.hash(session);}/*** @param session* @param userId* @Description 建立连接* @Throws* @Return void* @Date 2023-08-14 17:52:08* @Author WangKun*/@SneakyThrows@OnOpenpublic void onOpen(final Session session, @PathParam("userId") String userId) {this.session = session;this.userId = userId;session.setMaxIdleTimeout(SESSION_TIMEOUT);//先查找是否有uniCodeCopyOnWriteArraySet<WebSocket> users = WEB_SOCKET_MAP.get(userId);if (users == null) {//处理多个同时连接并发synchronized (WEB_SOCKET_MAP) {if (!WEB_SOCKET_MAP.contains(userId)) {users = new CopyOnWriteArraySet<>();WEB_SOCKET_MAP.put(userId, users);}}}users.add(this);sendMessage(String.valueOf(ResponseCode.CONNECT_SUCCESS.getCode()));log.info("用户--->{} 连接成功,当前在线人数为--->{}", userId, WEB_SOCKET_MAP.size());}/*** @param message* @Description 向客户端发送消息 session.getBasicRemote()与session.getAsyncRemote()的区别* @Throws* @Return void* @Date 2023-08-14 17:51:07* @Author WangKun*/@SneakyThrowspublic void sendMessage(String message) {// 加锁避免阻塞// 如果有多个客户端的话，亦或者同一个用户，或者打开了多个浏览器（同一个用户打开多个客户端或者多个界面），开了多个页面,此时Session是不同的,只能锁住一个session,所以锁住全局静态对象
// synchronized(session) {
// this.session.getBasicRemote().sendText(message);
// }synchronized (WEB_SOCKET_MAP) {CopyOnWriteArraySet<WebSocket> users = WEB_SOCKET_MAP.get(userId);if (users != null) {for (WebSocket user : users) {// 判断当前客户端的用户是否打开连接if (user.session.isOpen()) {user.session.getBasicRemote().sendText(message);log.info("向客户端发送数据--->{} 数据为--->{}", userId, message);}}}}}/*** @param* @Description 关闭连接* @Throws* @Return void* @Date 2023-08-14 17:52:30* @Author WangKun*/@SneakyThrows@OnClosepublic void onClose(Session session) {// 避免多人同时在线直接关闭通道。CopyOnWriteArraySet<WebSocket> copyOnWriteArraySet = WEB_SOCKET_MAP.get(this.userId);if (!copyOnWriteArraySet.isEmpty()) {Object[] objects = copyOnWriteArraySet.toArray();for (Object object : objects) {if (((WebSocket) object).session.equals(session)) {//删除当前用户WEB_SOCKET_MAP.get(this.userId).remove((WebSocket) object);// 如果有一个客户端登陆 下线清除用户if (WEB_SOCKET_MAP.get(this.userId).isEmpty()) {WEB_SOCKET_MAP.remove(this.userId);}CloseReason close = new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "关闭客户端，下线！");session.close(close);log.info("用户--->{} 关闭连接！", userId);}}}}/*** @param message* @param session* @Description 收到客户端消息* @Throws* @Return void* @Date 2023-08-15 10:54:55* @Author WangKun*/@SneakyThrows@OnMessagepublic void onMessage(String message, Session session) {//枷锁避免多个资源互抢//这一块可以操作数据，比如存到数据// 同一个用户，多个地方登录（多个session），循环发送消息,// 如果有多个客户端的话，亦或者同一个用户，或者打开了多个浏览器，开了多个页面,此时Session是不同的,只能锁住一个session,所以锁住全局静态对象synchronized (WEB_SOCKET_MAP) {CopyOnWriteArraySet<WebSocket> users = WEB_SOCKET_MAP.get(userId);if (users != null) {for (WebSocket user : users) {if (user.session.isOpen()) {user.session.getBasicRemote().sendText("pong");log.info("收到客户端发送的心跳的数据--->{} 数据为--->{}", userId, message);}}}}}/*** @param session* @param error* @Description 发生错误时* @Throws* @Return void* @Date 2023-08-15 10:55:27* @Author WangKun*/@SneakyThrows@OnErrorpublic void onError(Session session, Throwable error) {CopyOnWriteArraySet<WebSocket> users = WEB_SOCKET_MAP.get(userId);if (!users.isEmpty()) {Object[] objects = users.toArray();for (Object object : objects) {if (((WebSocket) object).session.equals(session)) {//删除当前用户WEB_SOCKET_MAP.get(this.userId).remove((WebSocket) object);// 如果有一个客户端登陆 下线清除用户if (WEB_SOCKET_MAP.get(this.userId).isEmpty()) {WEB_SOCKET_MAP.remove(this.userId);}CloseReason close = new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, "异常，下线！");session.close(close);log.error("用户--->{} 错误！" + userId, "原因--->{}" + error.getMessage(), error);}}}
// WEB_SOCKET_MAP.remove(userId);
// log.error("用户--->{} 错误！" + userId, "原因--->{}" + error.getMessage(), error);}/*** @param userId* @param message* @Description 通过userId向客户端发送消息（指定用户发送）* @Throws* @Return void* @Date 2023-08-14 18:01:35* @Author WangKun*/public static void sendTextMessageByUserId(String userId, String message) {CopyOnWriteArraySet<WebSocket> users = WEB_SOCKET_MAP.get(userId);if (users != null) {for (WebSocket user : users) {user.sendMessage(message);log.info("服务端发送消息到用户{},消息：{}", userId, message);}}}/*** @param message* @Description 群发自定义消息* @Throws* @Return void* @Date 2023-08-14 18:03:38* @Author WangKun*/public static void sendTextMessage(String message) {// 如果在线一个就广播if (!WEB_SOCKET_MAP.isEmpty()) {for (String item : WEB_SOCKET_MAP.keySet()) {CopyOnWriteArraySet<WebSocket> users = WEB_SOCKET_MAP.get(item);if (users != null) {for (WebSocket user : users) {user.sendMessage(message);log.info("服务端发送消息到用户{},消息：{}", item, message);}}}}}
}