一、WebSocket 基础知识
二、若依框架集成WebSocket
- RUOYI-VUE 集成websocket后 使用SecurityUtils获取用户信息报错 · Issue #I49SKT · 若依/RuoYi - Gitee.com
- webSocket接口中无法使用@Autowired和@Resource自动装配 · Issue #I5VT2N · 若依/RuoYi-Vue - Gitee.com
- 请问vue websocket握手如何在请求头中携带token · Issue #I5KAKK · 若依/RuoYi-Cloud - Gitee.com
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* websocket 配置
*
* @author ruoyi
*/
@Configuration
public class WebSocketConfig
{
@Bean
public ServerEndpointExporter serverEndpointExporter()
{
return new ServerEndpointExporter();
}
}
工具类 SemaphoreUtils
import java.util.concurrent.Semaphore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 信号量相关处理
*
* @author ruoyi
*/
public class SemaphoreUtils{
/**
* SemaphoreUtils 日志控制器
*/
private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreUtils.class);
/**
* 获取信号量
*
* @param semaphore
* @return
*/
public static boolean tryAcquire(Semaphore semaphore)
{
boolean flag = false;
try
{
flag = semaphore.tryAcquire();
}
catch (Exception e)
{
LOGGER.error("获取信号量异常", e);
}
return flag;
}
/**
* 释放信号量
*
* @param semaphore
*/
public static void release(Semaphore semaphore)
{
try
{
semaphore.release();
}
catch (Exception e)
{
LOGGER.error("释放信号量异常", e);
}
}
}
服务端类WebSocketServer
import java.util.concurrent.Semaphore;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import com.lxh.demo.util.SemaphoreUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* websocket 消息处理
*
* @author ruoyi
*/
@Component
@ServerEndpoint("/websocket/message")
public class WebSocketServer
{
/**
* WebSocketServer 日志控制器
*/
private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);
/**
* 默认最多允许同时在线人数100
*/
public static int socketMaxOnlineCount = 100;
private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);
/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(Session session) throws Exception{
boolean semaphoreFlag = false;
// 尝试获取信号量
semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
if (!semaphoreFlag)
{
// 未获取到信号量
LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数:" + socketMaxOnlineCount);
session.close();
}
else
{
// 添加用户
WebSocketUsers.put(session.getId(), session);
LOGGER.info("\n 建立连接 - {}", session);
LOGGER.info("\n 当前人数 - {}", WebSocketUsers.getUsers().size());
WebSocketUsers.sendMessageToUserByText(session, "连接成功");
}
}
/**
* 连接关闭时处理
*/
@OnClose
public void onClose(Session session)
{
LOGGER.info("\n 关闭连接 - {}", session);
// 移除用户
WebSocketUsers.remove(session.getId());
// 获取到信号量则需释放
SemaphoreUtils.release(socketSemaphore);
}
/**
* 抛出异常时处理
*/
@OnError
public void onError(Session session, Throwable exception) throws Exception
{
if (session.isOpen())
{
// 关闭连接
session.close();
}
String sessionId = session.getId();
LOGGER.info("\n 连接异常 - {}", sessionId);
LOGGER.info("\n 异常信息 - {}", exception);
// 移出用户
WebSocketUsers.remove(sessionId);
// 获取到信号量则需释放
SemaphoreUtils.release(socketSemaphore);
}
/**
* 服务器接收到客户端消息时调用的方法
*/
@OnMessage
public void onMessage(String message, Session session)
{
String msg = message.replace("你", "我").replace("吗", "");
WebSocketUsers.sendMessageToUserByText(session, msg);
}
}
WebSocketUsers工具类
import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import javax.websocket.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* websocket 客户端用户集
*
* @author ruoyi
*/
public class WebSocketUsers
{
/**
* WebSocketUsers 日志控制器
*/
private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);
/**
* 用户集
*/
private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();
/**
* 存储用户
*
* @param key 唯一键
* @param session 用户信息
*/
public static void put(String key, Session session)
{
USERS.put(key, session);
}
/**
* 移除用户
*
* @param session 用户信息
*
* @return 移除结果
*/
public static boolean remove(Session session)
{
String key = null;
boolean flag = USERS.containsValue(session);
if (flag)
{
Set<Map.Entry<String, Session>> entries = USERS.entrySet();
for (Map.Entry<String, Session> entry : entries)
{
Session value = entry.getValue();
if (value.equals(session))
{
key = entry.getKey();
break;
}
}
}
else
{
return true;
}
return remove(key);
}
/**
* 移出用户
*
* @param key 键
*/
public static boolean remove(String key)
{
LOGGER.info("\n 正在移出用户 - {}", key);
Session remove = USERS.remove(key);
if (remove != null)
{
boolean containsValue = USERS.containsValue(remove);
LOGGER.info("\n 移出结果 - {}", containsValue ? "失败" : "成功");
return containsValue;
}
else
{
return true;
}
}
/**
* 获取在线用户列表
*
* @return 返回用户集合
*/
public static Map<String, Session> getUsers()
{
return USERS;
}
/**
* 群发消息文本消息
*
* @param message 消息内容
*/
public static void sendMessageToUsersByText(String message)
{
Collection<Session> values = USERS.values();
for (Session value : values)
{
sendMessageToUserByText(value, message);
}
}
/**
* 发送文本消息
*
* @param session 缓存
* @param message 消息内容
*/
public static void sendMessageToUserByText(Session session, String message)
{
if (session != null)
{
try
{
session.getBasicRemote().sendText(message);
}
catch (IOException e)
{
LOGGER.error("\n[发送消息异常]", e);
}
}
else
{
LOGGER.info("\n[你已离线]");
}
}
}
Html 页面代码
<!DOCTYPE html>
<html lang="zh" xmlns:th="http://www.thymeleaf.org">
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<title>测试界面</title>
</head>
<body>
<div>
<input type="text" style="width: 20%" value="ws://127.0.0.1/websocket/message" id="url">
<button id="btn_join">连接</button>
<button id="btn_exit">断开</button>
</div>
<br/>
<textarea id="message" cols="100" rows="9"></textarea> <button id="btn_send">发送消息</button>
<br/>
<br/>
<textarea id="text_content" readonly="readonly" cols="100" rows="9"></textarea>返回内容
<br/>
<br/>
<script th:src="@{/js/jquery.min.js}" ></script>
<script type="text/javascript">
$(document).ready(function(){
var ws = null;
// 连接
$('#btn_join').click(function() {
var url = $("#url").val();
ws = new WebSocket(url);
ws.onopen = function(event) {
$('#text_content').append('已经打开连接!' + '\n');
}
ws.onmessage = function(event) {
$('#text_content').append(event.data + '\n');
}
ws.onclose = function(event) {
$('#text_content').append('已经关闭连接!' + '\n');
}
});
// 发送消息
$('#btn_send').click(function() {
var message = $('#message').val();
if (ws) {
ws.send(message);
} else {
alert("未连接到服务器");
}
});
//断开
$('#btn_exit').click(function() {
if (ws) {
ws.close();
ws = null;
}
});
})
</script>
</body>
</html>
成功运行后,页面如下
注意此时没有走用户认证,那么就要对路径放行,因为若依框架用的是SpringSecurity,所以找到文件SecurityConfig.java ,进行路径放行
三、用户认证问题
虽然按着上述步骤我们完成了浏览器(客户端)和Java(服务端)的WebSocket通信,但是我们不能限定哪些用户可以连接我们的服务端获取数据,服务端也不知道应该具体给哪些用户发送消息,在我们框架之前交互我们是通过浏览器传递toke 值来实现用户身份确认的,那么我们的WebSocket可不可以也这样呢?
很不幸的是 ws连接是无法像http一样完全自主定义请求头的,给token认证带来了不便,我们大致可以通过以下集中方式完成用户认证
1、将 token
明文携带在 url
中,例如ws://localhost:8080/weggo/websocket/message?Authorization=Bearer+token
2、通过websocket下的子协议来实现,Stomp这个协议来实现,前端采用SocketJs框架来实现对应定制请求头。实现携带authorization=Bearer +token 的需求,这样就可以正常建立连接
3、利用子协议数组,将 token 携带在 protocols 里,var ws = new WebSocket(url, ["token"]);
这样后端在 onOpen 事件中,就可以从 server 中读取 Sec-WebSocket-Protocol 属性来进行 token 的获取,具体可以参考WebScoket构造函数官方文档
var aWebSocket = new WebSocket(url [, protocols]);
url
要连接的URL;这应该是WebSocket服务器将响应的URL。
protocols 可选
一个协议字符串或者一个包含协议字符串的数组。这些字符串用于指定子协议,这样单个服务器可以实现多个WebSocket子协议
(例如,您可能希望一台服务器能够根据指定的协议(protocol)处理不同类型的交互)。如果不指定协议字符串,则假定为空字符串。
protocols对应的就是发起ws连接时, 携带在请求头中的Sec-WebSocket-Protocol属性, 服务端可以获取到此属性的值用于通信逻辑(即通信子协议,当然用来进行token认证也是完全没问题的),前端人员在请求头上携带sec-websocket-protocol=Bearer +token后台在请求到达oauth2之前进行拦截,然后将在请求头上添加Authorization=Bearer +token(key首字母大写),然后在响应头(respone)上添加sec-websocket-protocol=Bearer +token(不添加会报错)
方法3部分代码示例
//前端
var aWebSocket = new WebSocket(url ['用户token']);
//后端
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
//这里就是我们所提交的token
String submitedToken=session.getHandshakeHeaders().get("sec-websocket-protocol").get(0);
//根据token取得登录用户信息(业务逻辑根据你自己的来处理)
}
另外,如果需要在第一次握手前的时候就取得token,只需要在header里面取得就可以啦
@Override
public boolean beforeHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Map<String, Object> map) throws Exception {
System.out.println("准备握手");
String submitedToken = serverHttpRequest.getHeaders().get("sec-websocket-protocol")
return true;
}
因为我的项目是APP 移动端与服务端进行交互,所以后来选择了最简单实现的方案一
首先要解决的就是在拦截器获取url 的token 信息,原框架只从head里面获取,所以需要稍加改动
找到TokenService.java文件里的getToken方法,改成如下,这样就可以获取url 中的token 了又不影响原来的Http 请求
private String getToken(HttpServletRequest request)
{
String token = Optional.ofNullable(request.getHeader(header)).orElse(request.getParameter(header));
if (StringUtils.isNotEmpty(token) && token.startsWith(Constants.TOKEN_PREFIX))
{
token = token.replace(Constants.TOKEN_PREFIX, "");
}
return token;
}
接下来就是需要对我们的WebSocket类进行改造了,为了方便阅读,去除了WebSocketUsers类,添加了类变量webSocketSet来存储客户端对象文章来源:https://www.toymoban.com/news/detail-733263.html
import com.alibaba.fastjson2.JSON;
import com.tongchuang.common.utils.SecurityUtils;
import com.tongchuang.web.mqtt.domain.DeviceInfo;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
/**
* websocket 消息处理
*
* @author stronger
*/
@Component
@ServerEndpoint("/websocket/message")
public class WebSocketServer {
/*========================声明类变量,意在所有实例共享=================================================*/
/**
* WebSocketServer 日志控制器
*/
private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);
/**
* 默认最多允许同时在线人数100
*/
public static int socketMaxOnlineCount = 100;
private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);
HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.SECONDS, 8);
/**
* concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。
*/
private static final CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();
/**
* 连接数
*/
private static final AtomicInteger count = new AtomicInteger();
/*========================声明实例变量,意在每个实例独享=======================================================*/
/**
* 与某个客户端的连接会话,需要通过它来给客户端发送数据
*/
private Session session;
/**
* 用户id
*/
private String sid = "";
/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(Session session) throws Exception {
// 尝试获取信号量
boolean semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
if (!semaphoreFlag) {
// 未获取到信号量
LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
// 给当前Session 登录用户发送消息
sendMessageToUserByText(session, "当前在线人数超过限制数:" + socketMaxOnlineCount);
session.close();
} else {
// 返回此会话的经过身份验证的用户,如果此会话没有经过身份验证的用户,则返回null
Authentication authentication = (Authentication) session.getUserPrincipal();
SecurityUtils.setAuthentication(authentication);
String username = SecurityUtils.getUsername();
this.session = session;
//如果存在就先删除一个,防止重复推送消息
for (WebSocketServer webSocket : webSocketSet) {
if (webSocket.sid.equals(username)) {
webSocketSet.remove(webSocket);
count.getAndDecrement();
}
}
count.getAndIncrement();
webSocketSet.add(this);
this.sid = username;
LOGGER.info("\n 当前人数 - {}", count);
sendMessageToUserByText(session, "连接成功");
}
}
/**
* 连接关闭时处理
*/
@OnClose
public void onClose(Session session) {
LOGGER.info("\n 关闭连接 - {}", session);
// 移除用户
webSocketSet.remove(session);
// 获取到信号量则需释放
SemaphoreUtils.release(socketSemaphore);
}
/**
* 抛出异常时处理
*/
@OnError
public void onError(Session session, Throwable exception) throws Exception {
if (session.isOpen()) {
// 关闭连接
session.close();
}
String sessionId = session.getId();
LOGGER.info("\n 连接异常 - {}", sessionId);
LOGGER.info("\n 异常信息 - {}", exception);
// 移出用户
webSocketSet.remove(session);
// 获取到信号量则需释放
SemaphoreUtils.release(socketSemaphore);
}
/**
* 服务器接收到客户端消息时调用的方法
*/
@OnMessage
public void onMessage(String message, Session session) {
Authentication authentication = (Authentication) session.getUserPrincipal();
LOGGER.info("收到来自" + sid + "的信息:" + message);
// 实时更新
this.refresh(sid, authentication);
sendMessageToUserByText(session, "我收到了你的新消息哦");
}
/**
* 刷新定时任务,发送信息
*/
private void refresh(String userId, Authentication authentication) {
this.start(5000L, task -> {
// 判断用户是否在线,不在线则不用处理,因为在内部无法关闭该定时任务,所以通过返回值在外部进行判断。
if (WebSocketServer.isConn(userId)) {
// 因为这里是长链接,不会和普通网页一样,每次发送http 请求可以走拦截器【doFilterInternal】续约,所以需要手动续约
SecurityUtils.setAuthentication(authentication);
// 从数据库或者缓存中获取信息,构建自定义的Bean
DeviceInfo deviceInfo = DeviceInfo.builder().Macaddress("de5a735951ee").Imei("351517175516665")
.Battery("99").Charge("0").Latitude("116.402649").Latitude("39.914859").Altitude("80")
.Method(SecurityUtils.getUsername()).build();
// TODO判断数据是否有更新
// 发送最新数据给前端
WebSocketServer.sendInfo("JSON", deviceInfo, userId);
// 设置返回值,判断是否需要继续执行
return true;
}
return false;
});
}
private void start(long delay, Function<Timeout, Boolean> function) {
timer.newTimeout(t -> {
// 获取返回值,判断是否执行
Boolean result = function.apply(t);
if (result) {
timer.newTimeout(t.task(), delay, TimeUnit.MILLISECONDS);
}
}, delay, TimeUnit.MILLISECONDS);
}
/**
* 判断是否有链接
*
* @return
*/
public static boolean isConn(String sid) {
for (WebSocketServer item : webSocketSet) {
if (item.sid.equals(sid)) {
return true;
}
}
return false;
}
/**
* 群发自定义消息
* 或者指定用户发送消息
*/
public static void sendInfo(String type, Object data, @PathParam("sid") String sid) {
// 遍历WebSocketServer对象集合,如果符合条件就推送
for (WebSocketServer item : webSocketSet) {
try {
//这里可以设定只推送给这个sid的,为null则全部推送
if (sid == null) {
item.sendMessage(type, data);
} else if (item.sid.equals(sid)) {
item.sendMessage(type, data);
}
} catch (IOException ignored) {
}
}
}
/**
* 实现服务器主动推送
*/
private void sendMessage(String type, Object data) throws IOException {
Map<String, Object> result = new HashMap<>();
result.put("type", type);
result.put("data", data);
this.session.getAsyncRemote().sendText(JSON.toJSONString(result));
}
/**
* 实现服务器主动推送-根据session
*/
public static void sendMessageToUserByText(Session session, String message) {
if (session != null) {
try {
session.getBasicRemote().sendText(message);
} catch (IOException e) {
LOGGER.error("\n[发送消息异常]", e);
}
} else {
LOGGER.info("\n[你已离线]");
}
}
}
文章来源地址https://www.toymoban.com/news/detail-733263.html
到了这里,关于若依框架集成WebSocket带用户信息认证的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!