适用于单客户端,一个账号登陆一个客户端,登陆多个客户端会报错
The remote endpoint was in state [TEXT_FULL_WRITING] 文章来源:https://www.toymoban.com/news/detail-689135.html
这是因为此时的session是不同的,只能锁住一个session,解决此问题的方法把全局静态对象锁住,因为账号是唯一的文章来源地址https://www.toymoban.com/news/detail-689135.html
/**
* @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
*/
@Bean
public 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
*/
@Override
public 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);
}
@Override
public int hashCode() {
return Objects.hash(session);
}
/**
* @param session
* @param userId
* @Description 建立连接
* @Throws
* @Return void
* @Date 2023-08-14 17:52:08
* @Author WangKun
*/
@SneakyThrows
@OnOpen
public void onOpen(final Session session, @PathParam("userId") String userId) {
this.session = session;
this.userId = userId;
session.setMaxIdleTimeout(SESSION_TIMEOUT);
//先查找是否有uniCode
CopyOnWriteArraySet<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
*/
@SneakyThrows
public 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
@OnClose
public 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
@OnMessage
public 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
@OnError
public 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);
}
}
}
}
}
}
到了这里,关于SpringBoot2.0集成WebSocket,多客户端的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!