移除websocket
liu
1 year ago
parent
400b39e34f
commit
82c887ee8c
@ -1,27 +0,0 @@
|
|||||||
package com.supervision.websocket;
|
|
||||||
|
|
||||||
import cn.hutool.json.JSONUtil;
|
|
||||||
import com.supervision.constant.UserTokenConstant;
|
|
||||||
import com.supervision.websocket.dto.UserWebSocketDTO;
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
|
||||||
import org.springframework.data.redis.connection.Message;
|
|
||||||
import org.springframework.data.redis.connection.MessageListener;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
|
|
||||||
@Slf4j
|
|
||||||
@Component
|
|
||||||
public class KickUserListener implements MessageListener {
|
|
||||||
|
|
||||||
@Autowired
|
|
||||||
private UserResourceCheck userResourceCheck;
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void onMessage(Message message, byte[] pattern) {
|
|
||||||
String messageString = message.toString();
|
|
||||||
UserWebSocketDTO user = JSONUtil.toBean(messageString, UserWebSocketDTO.class);
|
|
||||||
log.info("Redis的Channel:{}收到踢用户{}下线消息", UserTokenConstant.KICK_CHANNEL, user.getUuid());
|
|
||||||
userResourceCheck.kickUser(user.getUuid());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,24 +0,0 @@
|
|||||||
package com.supervision.websocket;
|
|
||||||
|
|
||||||
import com.supervision.constant.UserTokenConstant;
|
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
|
||||||
import org.springframework.data.redis.listener.ChannelTopic;
|
|
||||||
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
public class RedisListener {
|
|
||||||
|
|
||||||
@Autowired
|
|
||||||
private KickUserListener kickUserListener;
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory) {
|
|
||||||
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
|
|
||||||
container.setConnectionFactory(redisConnectionFactory);
|
|
||||||
container.addMessageListener(kickUserListener, new ChannelTopic(UserTokenConstant.KICK_CHANNEL));
|
|
||||||
return container;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,84 +0,0 @@
|
|||||||
package com.supervision.websocket;
|
|
||||||
|
|
||||||
import cn.hutool.core.util.ObjectUtil;
|
|
||||||
import cn.hutool.core.util.StrUtil;
|
|
||||||
import cn.hutool.json.JSONUtil;
|
|
||||||
import com.baomidou.lock.annotation.Lock4j;
|
|
||||||
import com.supervision.constant.UserTokenConstant;
|
|
||||||
import com.supervision.exception.BusinessException;
|
|
||||||
import com.supervision.websocket.dto.KickDTO;
|
|
||||||
import com.supervision.websocket.dto.NoResourceDTO;
|
|
||||||
import com.supervision.websocket.dto.WebSocketLoginDTO;
|
|
||||||
import lombok.Data;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.springframework.beans.factory.annotation.Value;
|
|
||||||
|
|
||||||
import javax.websocket.*;
|
|
||||||
|
|
||||||
import org.springframework.data.redis.core.RedisTemplate;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
import java.io.IOException;
|
|
||||||
|
|
||||||
@Slf4j
|
|
||||||
@Component
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
public class UserResourceCheck {
|
|
||||||
|
|
||||||
@Value("${human.resourceMaxNumber}")
|
|
||||||
private String resourceNumber;
|
|
||||||
|
|
||||||
private final RedisTemplate<String, String> redisTemplate;
|
|
||||||
|
|
||||||
@Lock4j(name = "achieveDiagnoseResource")
|
|
||||||
public boolean achieveDiagnoseResource() {
|
|
||||||
long humanMaxNumber = Long.parseLong(resourceNumber);
|
|
||||||
long currentSize = redisTemplate.opsForHash().size(UserTokenConstant.USER_WEBSOCKET_CACHE);
|
|
||||||
// 如果小于数字人最大连接数,则可以连接
|
|
||||||
return currentSize < humanMaxNumber;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Lock4j(name = "achieveDiagnoseResourceAndOpenConnection")
|
|
||||||
public void achieveDiagnoseResourceAndOpenConnection(String uuid, String userId, Session session) {
|
|
||||||
// 如果小于数字人最大连接数,则可以连接
|
|
||||||
if (!achieveDiagnoseResource()) {
|
|
||||||
try {
|
|
||||||
session.getBasicRemote().sendText(JSONUtil.toJsonStr(new NoResourceDTO()));
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.error("发送消息失败", e);
|
|
||||||
}
|
|
||||||
throw new BusinessException("暂时没有资源,建立连接失败");
|
|
||||||
}
|
|
||||||
redisTemplate.opsForHash().put(UserTokenConstant.USER_WEBSOCKET_CACHE, uuid, JSONUtil.toJsonStr(new WebSocketLoginDTO(uuid, userId, session.getId())));
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* websocket保活接口,每5秒钟发送一次消息
|
|
||||||
*/
|
|
||||||
|
|
||||||
// 实现一个方法用于踢下线用户,走的是Redis的消息队列
|
|
||||||
public void kickUser(String uuid) {
|
|
||||||
log.info("尝试踢uuid:{}下线", uuid);
|
|
||||||
Session session = WebSocketSessionPool.SESSION_POOL.get(uuid);
|
|
||||||
// 只有不是忽略剔除的sessionId才可以踢下线
|
|
||||||
if (ObjectUtil.isNotEmpty(session)) {
|
|
||||||
try {
|
|
||||||
session.getBasicRemote().sendText(JSONUtil.toJsonStr(new KickDTO()));
|
|
||||||
session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "用户被踢下线"));
|
|
||||||
WebSocketSessionPool.SESSION_POOL.remove(uuid);
|
|
||||||
// 然后从redis中移除掉
|
|
||||||
redisTemplate.opsForHash().delete(UserTokenConstant.USER_WEBSOCKET_CACHE, uuid);
|
|
||||||
log.info("踢UUID:{},sessionId:{} 下线成功", uuid, session.getId());
|
|
||||||
return;
|
|
||||||
} catch (IOException e) {
|
|
||||||
log.error("用户:{}的websocket连接异常", e.getMessage());
|
|
||||||
// TODO 如果用户连接异常,怎么办
|
|
||||||
|
|
||||||
}
|
|
||||||
}
|
|
||||||
log.info("踢UUID:{}下线,未找到用户,踢下线失败", uuid);
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
}
|
|
||||||
@ -1,37 +0,0 @@
|
|||||||
package com.supervision.websocket;
|
|
||||||
|
|
||||||
import cn.hutool.core.util.ObjectUtil;
|
|
||||||
import cn.hutool.json.JSONUtil;
|
|
||||||
import com.supervision.websocket.UserResourceCheck;
|
|
||||||
import com.supervision.websocket.WebSocketSessionPool;
|
|
||||||
import com.supervision.websocket.dto.KeepaliveDTO;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.springframework.scheduling.annotation.Scheduled;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
import javax.websocket.Session;
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.util.Map;
|
|
||||||
|
|
||||||
@Component
|
|
||||||
@Slf4j
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
public class WebsocketKeepaliveTask {
|
|
||||||
@Scheduled(fixedDelay = 5 * 1000L)
|
|
||||||
public void keepalive() {
|
|
||||||
for (Map.Entry<String, Session> entries : WebSocketSessionPool.SESSION_POOL.entrySet()) {
|
|
||||||
String userId = entries.getKey();
|
|
||||||
Session session = entries.getValue();
|
|
||||||
if (ObjectUtil.isNotEmpty(session)) {
|
|
||||||
try {
|
|
||||||
session.getBasicRemote().sendText(JSONUtil.toJsonStr(new KeepaliveDTO()));
|
|
||||||
log.info("用户:{}的websocket保活成功", userId);
|
|
||||||
} catch (IOException e) {
|
|
||||||
log.error("用户:{}的websocket连接异常", userId, e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
@ -1,12 +0,0 @@
|
|||||||
package com.supervision.websocket.dto;
|
|
||||||
|
|
||||||
import com.supervision.constant.UserTokenConstant;
|
|
||||||
import lombok.Data;
|
|
||||||
|
|
||||||
@Data
|
|
||||||
public class KeepaliveDTO {
|
|
||||||
|
|
||||||
private final Integer code = UserTokenConstant.KEEPALIVE_CODE;
|
|
||||||
|
|
||||||
private final String message = "keepalive";
|
|
||||||
}
|
|
||||||
@ -1,12 +0,0 @@
|
|||||||
package com.supervision.websocket.dto;
|
|
||||||
|
|
||||||
import com.supervision.constant.UserTokenConstant;
|
|
||||||
import lombok.Data;
|
|
||||||
|
|
||||||
@Data
|
|
||||||
public class KickDTO {
|
|
||||||
|
|
||||||
private final Integer code = UserTokenConstant.KICK_CODE;
|
|
||||||
|
|
||||||
private final String message = "用户被踢下线";
|
|
||||||
}
|
|
||||||
@ -1,10 +0,0 @@
|
|||||||
package com.supervision.websocket.dto;
|
|
||||||
|
|
||||||
import com.supervision.constant.UserTokenConstant;
|
|
||||||
|
|
||||||
public class NoResourceDTO {
|
|
||||||
|
|
||||||
private final Integer code = UserTokenConstant.NO_RESOURCE_CODE;
|
|
||||||
|
|
||||||
private final String message = "用户被踢下线";
|
|
||||||
}
|
|
||||||
@ -1,12 +0,0 @@
|
|||||||
package com.supervision.websocket.dto;
|
|
||||||
|
|
||||||
import lombok.AllArgsConstructor;
|
|
||||||
import lombok.Data;
|
|
||||||
|
|
||||||
@Data
|
|
||||||
@AllArgsConstructor
|
|
||||||
public class UserWebSocketDTO {
|
|
||||||
|
|
||||||
private String uuid;
|
|
||||||
|
|
||||||
}
|
|
||||||
@ -1,28 +0,0 @@
|
|||||||
package com.supervision.websocket.dto;
|
|
||||||
|
|
||||||
import lombok.Data;
|
|
||||||
|
|
||||||
import java.time.LocalDateTime;
|
|
||||||
import java.time.format.DateTimeFormatter;
|
|
||||||
|
|
||||||
@Data
|
|
||||||
public class WebSocketLoginDTO {
|
|
||||||
|
|
||||||
private String userId;
|
|
||||||
|
|
||||||
private String uuid;
|
|
||||||
|
|
||||||
private String sessionId;
|
|
||||||
|
|
||||||
private String loginTime;
|
|
||||||
|
|
||||||
public WebSocketLoginDTO(String uuid, String userId, String sessionId) {
|
|
||||||
this.userId = userId;
|
|
||||||
this.uuid = uuid;
|
|
||||||
this.sessionId = sessionId;
|
|
||||||
this.loginTime = LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
|
|
||||||
}
|
|
||||||
|
|
||||||
public WebSocketLoginDTO() {
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Loading…
Reference in New Issue