项目中需要对应数据表有新数据入库的时候通知用户，不同的用户角色需要推送的通知不同。
集成、需求开发本文是基于Spring封装的方式集成，简单集成可以使用原生注解，更加方便简洁。
依赖引入项目使用的gradle
implementation 'org.springframework.boot:spring-boot-starter-websocket'session管理
由于需求问题，需要对session以包含的关系进行一层包装。
@Data
@NoArgsConstructor
@AllArgsConstructor
public class RoleWebSocketSession {
private WebSocketSession webSocketSession;
private String userType;
}
会话管理器，用ConcurrentHashMap来储存会话，key是jwt token字符串，value是包装过的RoleWebSocketSession 。
public class WsSessionManager {
//保存连接 session 的会话池
private static ConcurrentHashMap SESSION_POOL = new ConcurrentHashMap<>();
public static RoleWebSocketSession add(String key, RoleWebSocketSession session) {
// 添加 session
return SESSION_POOL.put(key, session);
}
public static RoleWebSocketSession remove(String key) {
return SESSION_POOL.remove(key);
}
public static void removeAndClose(String key) {
RoleWebSocketSession session = remove(key);
if (session != null) {
try {
// 关闭连接
session.getWebSocketSession().close();
} catch (IOException e) {
// 关闭出现异常处理
e.printStackTrace();
}
}
}
public static RoleWebSocketSession getRoleWebSocketSession(String key) {
return SESSION_POOL.get(key);
}
public static WebSocketSession getWebSocketSession(String key) {
// 获得 session
return SESSION_POOL.get(key).getWebSocketSession();
}
public static Integer getSessionPoolSize() {
return SESSION_POOL.size();
}
public static synchronized void sendMsg(String key, String msg) {
WebSocketMessage webSocketMessage = new TextMessage(msg);
try {
getWebSocketSession(key).sendMessage(webSocketMessage);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
public static void sendMsgToSpecificRoles(Set roles, String msg) {
SESSION_POOL.forEach((token, session) -> {
if (roles.contains(session.getUserType())) {
sendMsg(token, msg);
}
});
}
public static void sendMsgToSpecificRole(String role, String msg) {
SESSION_POOL.forEach((token, session) -> {
if (role.equals(session.getUserType())) {
sendMsg(token, msg);
}
});
}
public static void sendMsgToAll(String msg) {
SESSION_POOL.forEach((token, session) -> {
sendMsg(token, msg);
});
}
}
WebSocket处理器
就是一个WebSocket的endpoint（终端），相当于原生注解的@ServerEndpoint("/InfoPush")标记的类，对WebSocket的事件进行处理。
@Component
@Slf4j
public class InfoPushHandler extends TextWebSocketHandler {
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
Object token = session.getAttributes().get("token");
if (null != token) {
TokenData tokenDataValue = JwtUtil.getTokenDataValue(token.toString());
if (tokenDataValue.isValid()) {
WsSessionManager.add(token.toString(), new RoleWebSocketSession(session, tokenDataValue.getUserType()));
log.info("[WebScoket]:用户" + tokenDataValue.getUsername() + "连接成功,现有连接数:" + WsSessionManager.getSessionPoolSize());
}else {
throw new RuntimeException("[WebScoket]:用户token已经失效!");
}
}else {
throw new RuntimeException("[WebScoket]:用户token已经失效!");
}
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
Object token = session.getAttributes().get("token");
if (null != token) {
WsSessionManager.remove(token.toString());
TokenData tokenDataValue = JwtUtil.getTokenDataValue(token.toString());
log.info("[WebScoket]:用户" + tokenDataValue.getUsername() + "断开连接,现有连接数:" + WsSessionManager.getSessionPoolSize());
}
}
}
WebSocket握手拦截器
需要对jwt token有一些处理，WebSocket建立长连接之前会通过HTTP进行一次握手，具体过程不深入，在这个过程进行拦截处理是最节省资源的，因为握手检验没有通过直接拒绝连接，不会继续进行WebSocket长连接的建立。
@Component
@Slf4j
public class TokenInterceptor implements HandshakeInterceptor {
@Override
public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map attributes) throws Exception {
//HttpUtil工具类是Hutool的，需要的话自己引入
Map paramMap = HttpUtil.decodeParamMap(request.getURI().getQuery(), Charset.forName("UTF-8"));
//下面代码啊就是看业务需求了，仅供参考，直接复制的话，肯定是有类缺失的
String token = paramMap.get("token");
TokenData tokenDataValue;
if (null != token) {
tokenDataValue = JwtUtil.getTokenDataValue(token);
if (tokenDataValue.isValid()) {
attributes.put("token", token);
log.info("[WebScoket]:用户" + tokenDataValue.getUsername() + "握手");
return true;
}
log.info("[WebScoket]:用户" + tokenDataValue.getUsername() + "握手失败");
}
return false;
}
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
}
}
WebSocketConfig
WebSocket配置类，开启WebSocket，添加处理器和拦截器，配置允许跨域。
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
private final InfoPushHandler infoPushHandler;
private final TokenInterceptor tokenInterceptor;
@Autowired
public WebSocketConfig(InfoPushHandler infoPushHandler, TokenInterceptor tokenInterceptor) {
this.infoPushHandler = infoPushHandler;
this.tokenInterceptor = tokenInterceptor;
}
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
registry
.addHandler(infoPushHandler, "InfoPush")//对应ws的地址
.addInterceptors(tokenInterceptor)//添加拦截器
.setAllowedOrigins("*");//本地测试或者有需要可以开启跨域
}
}
到这里后端Spring Boot集成WebSocket就基本完成了
前端建立WebSocket连接图省事的话可以使用工具网站进行测试http://coolaf.com/zh/tool/chattest
这里前端使用的是vue
export default {
data() {
return {
ws: null,//建立的连接
}
},
created() {
this.initWs();
},
methods: {
initWs(){
//这里的url就是 ws:// + 项目的地址和端口还有ContentPath + 在WebSocket配置的Handler的path（注意大小写） + 参数拼接
//比如下面这个拼完的链接：ws://127.0.0.1:8080/wsDemo/InfoPush?token=xxxxxxxxxxxxxxxxxxxxxx
let url = dataInit.wsInfoPush + "?token=" + this.$store.state.login.token;
this.ws = new WebSocket(url);
this.ws.onopen = this.onopen;
this.ws.onmessage = this.onmessage;
this.ws.onclose = this.onclose;
this.ws.onerror = this.onerror;
},
onopen() {
console.log("websocket：连接成功");
},
onmessage(e) {
let msg = e.data;
console.log(msg);
},
onerror(e) {
console.log("websocket：出现错误");
this.ws.close();
this.ws = null;
},
onclose(e) {
console.log("websocket：连接关闭");
console.log('websocket：断开: ' + e.code + ' ' + e.reason + ' ' + e.wasClean);
},
},
}
集成开发到这里基本就全部完成了，下面就是我在这个过程中遇到的问题和解决方案，主要是关于Nginx的，仅供参考。
坑 前后端分离，前端部署在Nginx，通过Nginx代理连接WebSocket成功后，过一会自动断开原因：Ngixn的location中的proxy_read_timeout 默认60s，60s没有任何通讯会自动断开连接
解决方案：可以把这个时间设置大一些600s（十分钟）
location /wsDemo/InfoPush { #websocket转发
proxy_pass http://127.0.0.1:8080/wsDemo/InfoPush;
proxy_redirect off;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 600s;
proxy_http_version 1.1;
proxy_set_header X-Forwarded-Port $Server_port;
proxy_set_header X-Forwarded-Proto $scheme;
}
设置太长也不好，可以考虑在前端做个简单的心跳机制和重连机制
export default {
data() {
return {
ws: null,//建立的连接
heartTimer: null, //心跳计时器
reconTimer: null, //重连计时器
}
},
created() {
this.initWs();
},
methods: {
initWs(){
//这里的url就是 ws:// + 项目的地址和端口还有ContentPath + 在WebSocket配置的Handler的path（注意大小写） + 参数拼接
//比如下面这个拼完的链接：ws://127.0.0.1:8080/wsDemo/InfoPush?token=xxxxxxxxxxxxxxxxxxxxxx
let url = dataInit.wsInfoPush + "?token=" + this.$store.state.login.token;
this.ws = new WebSocket(url);
this.ws.onopen = this.onopen;
this.ws.onmessage = this.onmessage;
this.ws.onclose = this.onclose;
this.ws.onerror = this.onerror;
},
onopen() {
console.log("websocket：连接成功");
this.heartCheck();//开启心跳检测，每五分钟发送一个心跳包
},
onmessage(e) {
let msg = e.data;
console.log(msg);
},
onerror(e) {
console.log("websocket：出现错误");
this.ws.close();
this.ws = null;
},
onclose(e) {
console.log("websocket：连接关闭");
console.log('websocket：断开: ' + e.code + ' ' + e.reason + ' ' + e.wasClean);
clearInterval(this.heartTimer)
this.reconWs();
},
heartCheck() {
console.log("websocket：开启心跳检测")
this.heartTimer = setInterval(() => {
console.log("websocket：heartCheck");
this.ws.send("heartCheck");
}, 10 * 1000);
},
reconWs() {
console.log("websocket：尝试重连");
this.reconTimer = setTimeout(() => {
this.initWs();
}, 5 * 1000);
},
},
}
