提交websocket相关代码

dev_2.0.0
liu
parent 5736487568
commit 718f395719

@ -2,6 +2,8 @@ package com.supervision.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
@Configuration
@ -14,4 +16,13 @@ public class WebSocketConfig {
public ServerEndpointExporter serverEndpointExporter(){
return new ServerEndpointExporter();
}
@Bean
public TaskScheduler taskScheduler(){
ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
threadPoolTaskScheduler.setThreadNamePrefix("SockJS-");
threadPoolTaskScheduler.setPoolSize(Runtime.getRuntime().availableProcessors());
threadPoolTaskScheduler.setRemoveOnCancelPolicy(true);
return threadPoolTaskScheduler;
}
}

@ -0,0 +1,48 @@
package com.supervision.task;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.supervision.constant.UserTokenConstant;
import com.supervision.controller.WebSocketServer;
import com.supervision.usermanage.UserResourceCheck;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
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 {
private final RedisTemplate<String, String> redisTemplate;
@Scheduled(fixedDelay = 5 * 1000L)
public void keepalive() {
log.info("websocket保活接口开始,每5秒钟发送一次消息");
for (Map.Entry<String, Session> entries : WebSocketServer.SESSION_POOL.entrySet()) {
String userId = entries.getKey();
Session session = entries.getValue();
if (ObjectUtil.isNotEmpty(session)) {
try {
session.getBasicRemote().sendText(JSONUtil.toJsonStr(new UserResourceCheck.Keepalive()));
log.info("用户:{}的websocket保活成功", userId);
} catch (IOException e) {
log.error("用户:{}的websocket连接异常", userId, e);
// 连接异常的用户,移除
WebSocketServer.SESSION_POOL.remove(userId);
// 移除redis中该用户的缓存
redisTemplate.opsForHash().delete(UserTokenConstant.USER_WEBSOCKET_CACHE, userId);
}
}
}
log.info("websocket保活接口结束");
}
}

@ -57,28 +57,6 @@ public class UserResourceCheck {
/**
* websocket,5
*/
@Scheduled(fixedDelay = 5 * 1000L)
public void keepalive() {
log.info("websocket保活接口开始,每5秒钟发送一次消息");
for (Map.Entry<String, Session> entries : WebSocketServer.SESSION_POOL.entrySet()) {
String userId = entries.getKey();
Session session = entries.getValue();
if (ObjectUtil.isNotEmpty(session)) {
try {
session.getBasicRemote().sendText(JSONUtil.toJsonStr(new Keepalive()));
log.info("用户:{}的websocket保活成功", userId);
} catch (IOException e) {
log.error("用户:{}的websocket连接异常", userId, e);
// 连接异常的用户,移除
WebSocketServer.SESSION_POOL.remove(userId);
// 移除redis中该用户的缓存
redisTemplate.opsForHash().delete(UserTokenConstant.USER_WEBSOCKET_CACHE, userId);
}
}
}
log.info("websocket保活接口结束");
}
// 实现一个方法用于踢下线用户,走的是Redis的消息队列
public void kickUser(String userId, String ignoreSessionId) {

Loading…
Cancel
Save