dyf/fys-Multi-tenant
1
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fys

Browse Source
This commit is contained in:
daiyongfei
2025-06-27 10:23:57 +08:00
parent 15d7ef8771
commit b94549185c
774 changed files with 3543 additions and 3558 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/config/WebSocketConfig.java Normal file
View File

@ -0,0 +1,63 @@
package com.fuyuanshen.common.websocket.config;
import cn.hutool.core.util.StrUtil;
import com.fuyuanshen.common.websocket.config.properties.WebSocketProperties;
import com.fuyuanshen.common.websocket.handler.PlusWebSocketHandler;
import com.fuyuanshen.common.websocket.interceptor.PlusWebSocketInterceptor;
import com.fuyuanshen.common.websocket.listener.WebSocketTopicListener;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.server.HandshakeInterceptor;
/**
* WebSocket 配置
*
* @author zendwang
*/
@AutoConfiguration
@ConditionalOnProperty(value = "websocket.enabled", havingValue = "true")
@EnableConfigurationProperties(WebSocketProperties.class)
@EnableWebSocket
public class WebSocketConfig {
@Bean
public WebSocketConfigurer webSocketConfigurer(HandshakeInterceptor handshakeInterceptor,
WebSocketHandler webSocketHandler, WebSocketProperties webSocketProperties) {
// 如果WebSocket的路径为空，则设置默认路径为 "/websocket"
if (StrUtil.isBlank(webSocketProperties.getPath())) {
webSocketProperties.setPath("/websocket");
}
// 如果允许跨域访问的地址为空，则设置为 "*"，表示允许所有来源的跨域请求
if (StrUtil.isBlank(webSocketProperties.getAllowedOrigins())) {
webSocketProperties.setAllowedOrigins("*");
}
// 返回一个WebSocketConfigurer对象，用于配置WebSocket
return registry -> registry
// 添加WebSocket处理程序和拦截器到指定路径，设置允许的跨域来源
.addHandler(webSocketHandler, webSocketProperties.getPath())
.addInterceptors(handshakeInterceptor)
.setAllowedOrigins(webSocketProperties.getAllowedOrigins());
}
@Bean
public HandshakeInterceptor handshakeInterceptor() {
return new PlusWebSocketInterceptor();
}
@Bean
public WebSocketHandler webSocketHandler() {
return new PlusWebSocketHandler();
}
@Bean
public WebSocketTopicListener topicListener() {
return new WebSocketTopicListener();
}
}

26
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/config/properties/WebSocketProperties.java Normal file
View File

@ -0,0 +1,26 @@
package com.fuyuanshen.common.websocket.config.properties;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* WebSocket 配置项
*
* @author zendwang
*/
@ConfigurationProperties("websocket")
@Data
public class WebSocketProperties {
private Boolean enabled;
/**
* 路径
*/
private String path;
/**
* 设置访问源地址
*/
private String allowedOrigins;
}

29
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/constant/WebSocketConstants.java Normal file
View File

@ -0,0 +1,29 @@
package com.fuyuanshen.common.websocket.constant;
/**
* websocket的常量配置
*
* @author zendwang
*/
public interface WebSocketConstants {
/**
* websocketSession中的参数的key
*/
String LOGIN_USER_KEY = "loginUser";
/**
* 订阅的频道
*/
String WEB_SOCKET_TOPIC = "global:websocket";
/**
* 前端心跳检查的命令
*/
String PING = "ping";
/**
* 服务端心跳恢复的字符串
*/
String PONG = "pong";
}

29
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/dto/WebSocketMessageDto.java Normal file
View File

@ -0,0 +1,29 @@
package com.fuyuanshen.common.websocket.dto;
import lombok.Data;
import java.io.Serial;
import java.io.Serializable;
import java.util.List;
/**
* 消息的dto
*
* @author zendwang
*/
@Data
public class WebSocketMessageDto implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
/**
* 需要推送到的session key 列表
*/
private List<Long> sessionKeys;
/**
* 需要发送的消息
*/
private String message;
}

122
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/handler/PlusWebSocketHandler.java Normal file
View File

@ -0,0 +1,122 @@
package com.fuyuanshen.common.websocket.handler;
import cn.hutool.core.util.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import com.fuyuanshen.common.core.domain.model.LoginUser;
import com.fuyuanshen.common.websocket.dto.WebSocketMessageDto;
import com.fuyuanshen.common.websocket.holder.WebSocketSessionHolder;
import com.fuyuanshen.common.websocket.utils.WebSocketUtils;
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import java.io.IOException;
import java.util.List;
import static com.fuyuanshen.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
/**
* WebSocketHandler 实现类
*
* @author zendwang
*/
@Slf4j
public class PlusWebSocketHandler extends AbstractWebSocketHandler {
/**
* 连接成功后
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) throws IOException {
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
if (ObjectUtil.isNull(loginUser)) {
session.close(CloseStatus.BAD_DATA);
log.info("[connect] invalid token received. sessionId: {}", session.getId());
return;
}
WebSocketSessionHolder.addSession(loginUser.getUserId(), session);
log.info("[connect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType());
}
/**
* 处理接收到的文本消息
*
* @param session WebSocket会话
* @param message 接收到的文本消息
* @throws Exception 处理消息过程中可能抛出的异常
*/
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
// 从WebSocket会话中获取登录用户信息
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
// 创建WebSocket消息DTO对象
WebSocketMessageDto webSocketMessageDto = new WebSocketMessageDto();
webSocketMessageDto.setSessionKeys(List.of(loginUser.getUserId()));
webSocketMessageDto.setMessage(message.getPayload());
WebSocketUtils.publishMessage(webSocketMessageDto);
}
/**
* 处理接收到的二进制消息
*
* @param session WebSocket会话
* @param message 接收到的二进制消息
* @throws Exception 处理消息过程中可能抛出的异常
*/
@Override
protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
super.handleBinaryMessage(session, message);
}
/**
* 处理接收到的Pong消息（心跳监测）
*
* @param session WebSocket会话
* @param message 接收到的Pong消息
* @throws Exception 处理消息过程中可能抛出的异常
*/
@Override
protected void handlePongMessage(WebSocketSession session, PongMessage message) throws Exception {
WebSocketUtils.sendPongMessage(session);
}
/**
* 处理WebSocket传输错误
*
* @param session WebSocket会话
* @param exception 发生的异常
* @throws Exception 处理过程中可能抛出的异常
*/
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
log.error("[transport error] sessionId: {} , exception:{}", session.getId(), exception.getMessage());
}
/**
* 在WebSocket连接关闭后执行清理操作
*
* @param session WebSocket会话
* @param status 关闭状态信息
*/
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
if (ObjectUtil.isNull(loginUser)) {
log.info("[disconnect] invalid token received. sessionId: {}", session.getId());
return;
}
WebSocketSessionHolder.removeSession(loginUser.getUserId());
log.info("[disconnect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType());
}
/**
* 指示处理程序是否支持接收部分消息
*
* @return 如果支持接收部分消息，则返回true；否则返回false
*/
@Override
public boolean supportsPartialMessages() {
return false;
}
}

74
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/holder/WebSocketSessionHolder.java Normal file
View File

@ -0,0 +1,74 @@
package com.fuyuanshen.common.websocket.holder;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketSession;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
* WebSocketSession 用于保存当前所有在线的会话信息
*
* @author zendwang
*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class WebSocketSessionHolder {
private static final Map<Long, WebSocketSession> USER_SESSION_MAP = new ConcurrentHashMap<>();
/**
* 将WebSocket会话添加到用户会话Map中
*
* @param sessionKey 会话键，用于检索会话
* @param session 要添加的WebSocket会话
*/
public static void addSession(Long sessionKey, WebSocketSession session) {
removeSession(sessionKey);
USER_SESSION_MAP.put(sessionKey, session);
}
/**
* 从用户会话Map中移除指定会话键对应的WebSocket会话
*
* @param sessionKey 要移除的会话键
*/
public static void removeSession(Long sessionKey) {
WebSocketSession session = USER_SESSION_MAP.remove(sessionKey);
try {
session.close(CloseStatus.BAD_DATA);
} catch (Exception ignored) {
}
}
/**
* 根据会话键从用户会话Map中获取WebSocket会话
*
* @param sessionKey 要获取的会话键
* @return 与给定会话键对应的WebSocket会话，如果不存在则返回null
*/
public static WebSocketSession getSessions(Long sessionKey) {
return USER_SESSION_MAP.get(sessionKey);
}
/**
* 获取存储在用户会话Map中所有WebSocket会话的会话键集合
*
* @return 所有WebSocket会话的会话键集合
*/
public static Set<Long> getSessionsAll() {
return USER_SESSION_MAP.keySet();
}
/**
* 检查给定的会话键是否存在于用户会话Map中
*
* @param sessionKey 要检查的会话键
* @return 如果存在对应的会话键，则返回true；否则返回false
*/
public static Boolean existSession(Long sessionKey) {
return USER_SESSION_MAP.containsKey(sessionKey);
}
}

75
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/interceptor/PlusWebSocketInterceptor.java Normal file
View File

@ -0,0 +1,75 @@
package com.fuyuanshen.common.websocket.interceptor;
import cn.dev33.satoken.exception.NotLoginException;
import cn.dev33.satoken.stp.StpUtil;
import lombok.extern.slf4j.Slf4j;
import com.fuyuanshen.common.core.domain.model.LoginUser;
import com.fuyuanshen.common.core.utils.ServletUtils;
import com.fuyuanshen.common.core.utils.StringUtils;
import com.fuyuanshen.common.satoken.utils.LoginHelper;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import java.util.Map;
import static com.fuyuanshen.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
/**
* WebSocket握手请求的拦截器
*
* @author zendwang
*/
@Slf4j
public class PlusWebSocketInterceptor implements HandshakeInterceptor {
/**
* WebSocket握手之前执行的前置处理方法
*
* @param request WebSocket握手请求
* @param response WebSocket握手响应
* @param wsHandler WebSocket处理程序
* @param attributes 与WebSocket会话关联的属性
* @return 如果允许握手继续进行，则返回true；否则返回false
*/
@Override
public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) {
try {
// 检查是否登录 是否有token
LoginUser loginUser = LoginHelper.getLoginUser();
// 解决 ws 不走 mvc 拦截器问题(cloud 版本不受影响)
// 检查 header 与 param 里的 clientid 与 token 里的是否一致
String headerCid = ServletUtils.getRequest().getHeader(LoginHelper.CLIENT_KEY);
String paramCid = ServletUtils.getParameter(LoginHelper.CLIENT_KEY);
String clientId = StpUtil.getExtra(LoginHelper.CLIENT_KEY).toString();
if (!StringUtils.equalsAny(clientId, headerCid, paramCid)) {
// token 无效
throw NotLoginException.newInstance(StpUtil.getLoginType(),
"-100", "客户端ID与Token不匹配",
StpUtil.getTokenValue());
}
attributes.put(LOGIN_USER_KEY, loginUser);
return true;
} catch (NotLoginException e) {
log.error("WebSocket 认证失败'{}',无法访问系统资源", e.getMessage());
return false;
}
}
/**
* WebSocket握手成功后执行的后置处理方法
*
* @param request WebSocket握手请求
* @param response WebSocket握手响应
* @param wsHandler WebSocket处理程序
* @param exception 握手过程中可能出现的异常
*/
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
// 在这个方法中可以执行一些握手成功后的后续处理逻辑，比如记录日志或者其他操作
}
}

50
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/listener/WebSocketTopicListener.java Normal file
View File

@ -0,0 +1,50 @@
package com.fuyuanshen.common.websocket.listener;
import cn.hutool.core.collection.CollUtil;
import lombok.extern.slf4j.Slf4j;
import com.fuyuanshen.common.websocket.holder.WebSocketSessionHolder;
import com.fuyuanshen.common.websocket.utils.WebSocketUtils;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
/**
* WebSocket 主题订阅监听器
*
* @author zendwang
*/
@Slf4j
public class WebSocketTopicListener implements ApplicationRunner, Ordered {
/**
* 在Spring Boot应用程序启动时初始化WebSocket主题订阅监听器
*
* @param args 应用程序参数
* @throws Exception 初始化过程中可能抛出的异常
*/
@Override
public void run(ApplicationArguments args) throws Exception {
// 订阅WebSocket消息
WebSocketUtils.subscribeMessage((message) -> {
log.info("WebSocket主题订阅收到消息session keys={} message={}", message.getSessionKeys(), message.getMessage());
// 如果key不为空就按照key发消息 如果为空就群发
if (CollUtil.isNotEmpty(message.getSessionKeys())) {
message.getSessionKeys().forEach(key -> {
if (WebSocketSessionHolder.existSession(key)) {
WebSocketUtils.sendMessage(key, message.getMessage());
}
});
} else {
WebSocketSessionHolder.getSessionsAll().forEach(key -> {
WebSocketUtils.sendMessage(key, message.getMessage());
});
}
});
log.info("初始化WebSocket主题订阅监听器成功");
}
@Override
public int getOrder() {
return -1;
}
}

127
fys-common/fys-common-websocket/src/main/java/com/fuyuanshen/common/websocket/utils/WebSocketUtils.java Normal file
View File

@ -0,0 +1,127 @@
package com.fuyuanshen.common.websocket.utils;
import cn.hutool.core.collection.CollUtil;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import com.fuyuanshen.common.redis.utils.RedisUtils;
import com.fuyuanshen.common.websocket.dto.WebSocketMessageDto;
import com.fuyuanshen.common.websocket.holder.WebSocketSessionHolder;
import org.springframework.web.socket.PongMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import static com.fuyuanshen.common.websocket.constant.WebSocketConstants.WEB_SOCKET_TOPIC;
/**
* 工具类
*
* @author zendwang
*/
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class WebSocketUtils {
/**
* 向指定的WebSocket会话发送消息
*
* @param sessionKey 要发送消息的用户id
* @param message 要发送的消息内容
*/
public static void sendMessage(Long sessionKey, String message) {
WebSocketSession session = WebSocketSessionHolder.getSessions(sessionKey);
sendMessage(session, message);
}
/**
* 订阅WebSocket消息主题，并提供一个消费者函数来处理接收到的消息
*
* @param consumer 处理WebSocket消息的消费者函数
*/
public static void subscribeMessage(Consumer<WebSocketMessageDto> consumer) {
RedisUtils.subscribe(WEB_SOCKET_TOPIC, WebSocketMessageDto.class, consumer);
}
/**
* 发布WebSocket订阅消息
*
* @param webSocketMessage 要发布的WebSocket消息对象
*/
public static void publishMessage(WebSocketMessageDto webSocketMessage) {
List<Long> unsentSessionKeys = new ArrayList<>();
// 当前服务内session,直接发送消息
for (Long sessionKey : webSocketMessage.getSessionKeys()) {
if (WebSocketSessionHolder.existSession(sessionKey)) {
WebSocketUtils.sendMessage(sessionKey, webSocketMessage.getMessage());
continue;
}
unsentSessionKeys.add(sessionKey);
}
// 不在当前服务内session,发布订阅消息
if (CollUtil.isNotEmpty(unsentSessionKeys)) {
WebSocketMessageDto broadcastMessage = new WebSocketMessageDto();
broadcastMessage.setMessage(webSocketMessage.getMessage());
broadcastMessage.setSessionKeys(unsentSessionKeys);
RedisUtils.publish(WEB_SOCKET_TOPIC, broadcastMessage, consumer -> {
log.info(" WebSocket发送主题订阅消息topic:{} session keys:{} message:{}",
WEB_SOCKET_TOPIC, unsentSessionKeys, webSocketMessage.getMessage());
});
}
}
/**
* 向所有的WebSocket会话发布订阅的消息(群发)
*
* @param message 要发布的消息内容
*/
public static void publishAll(String message) {
WebSocketMessageDto broadcastMessage = new WebSocketMessageDto();
broadcastMessage.setMessage(message);
RedisUtils.publish(WEB_SOCKET_TOPIC, broadcastMessage, consumer -> {
log.info("WebSocket发送主题订阅消息topic:{} message:{}", WEB_SOCKET_TOPIC, message);
});
}
/**
* 向指定的WebSocket会话发送Pong消息
*
* @param session 要发送Pong消息的WebSocket会话
*/
public static void sendPongMessage(WebSocketSession session) {
sendMessage(session, new PongMessage());
}
/**
* 向指定的WebSocket会话发送文本消息
*
* @param session WebSocket会话
* @param message 要发送的文本消息内容
*/
public static void sendMessage(WebSocketSession session, String message) {
sendMessage(session, new TextMessage(message));
}
/**
* 向指定的WebSocket会话发送WebSocket消息对象
*
* @param session WebSocket会话
* @param message 要发送的WebSocket消息对象
*/
private synchronized static void sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
if (session == null || !session.isOpen()) {
log.warn("[send] session会话已经关闭");
} else {
try {
session.sendMessage(message);
} catch (IOException e) {
log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
}
}
}
}