整体描述
在SpringBoot下使用websocket，达到前后端通信的目的，这里简单写下使用。就使用SpringBoot自带的websocket实现。websocket涉及的情况比较多，一定还有一些考虑不到的问题，这里只是提供一个思路，发现问题就具体问题具体分析了。
具体使用
1. 添加依赖
添加websocket的依赖，在pom里添加：
<!-- websocket 前后端通信-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
2. 消息格式
添加依赖之后，创建一个消息的结构体，也就是自定义消息，分为消息id，消息类型，消息内容和消息时间，前后端通信时，可以根据消息类型判断，对消息内容进行解析。比如消息类型为HeartBeat，也就是心跳，可能消息内容就是空，就不用对消息内容进行解析了，再比如消息类型是Broadcast，代表后台传给前端的广播信息，就需要对消息内容进行判断。
注：消息内容格式使用Json字符串，信息灵活多样。
import lombok.Data;
/**
* websocket消息体
*/
@Data
public class WebSocketMessage {
/**
* 消息ID
*/
private String messageId;
/**
* 消息类型
*/
private String messageType;
/**
* 消息内容
*/
private String messageContent;
/**
* 消息时间
*/
private String messageTime;
}
3. 具体操作
此类主要封装websocket相关操作，连接，发送消息和相关回调等。
@Component
@ServerEndpoint("/websocket/{username}") //暴露的ws应用的路径
public class WebSocket {
private static final Logger log = LoggerFactory.getLogger(WebSocket.class);
/**
* 超时时间间隔，超过此时间(单位:ms)，认为websocket客户端连接超时
* 和定时任务配合，定时任务在发送心跳一分钟之后检查各个Session
* 这里设定时间间隔为2分钟
*/
private static final Long timeInterval = (long) (2 * 60 * 1000);
/**
* 当前在线客户端数量
*/
private static AtomicInteger onlineClientNumber = new AtomicInteger(0);
/**
* 当前在线客户端集合
* 以键值对方式存储，key是连接的编号，value是连接的对象
*/
private static Map<String, Session> onlineClientMap = new ConcurrentHashMap<>();
/**
* 当前在线客户端时间集合，用于心跳判断
* 以键值对方式存储，key是连接的编号，value是最近一次通信时间
*/
private static Map<String, Long> onlineUserTimeMap = new ConcurrentHashMap<>();
/**
* 客户端与服务端连接成功
*/
@OnOpen
public void onOpen(Session session, @PathParam("username") String username) {
// 在线数+1
onlineClientNumber.incrementAndGet();
// 添加当前连接的session
onlineClientMap.put(session.getId(), session);
// 添加当前连接的session的时间
onlineUserTimeMap.put(session.getId(), System.currentTimeMillis());
log.info("onOpen---->time:[{}],User:[{}],Session:[{}],total:[{}]",
DateUtils.getTime(),
username,
session.getId(),
onlineClientNumber);
}
/**
* 客户端与服务端连接关闭
*/
@OnClose
public void onClose(Session session, @PathParam("username") String username) {
// 在线数-1
onlineClientNumber.decrementAndGet();
// 移除当前连接的session
onlineClientMap.remove(session.getId());
// 移除当前连接的session的时间
onlineUserTimeMap.remove(session.getId());
log.info("onClose---->Time:[{}],User:[{}],Session:[{}],total:[{}]",
DateUtils.getTime(),
username,
session.getId(),
onlineClientNumber);
}
/**
* 客户端与服务端连接异常
*/
@OnError
public void onError(Throwable error, Session session, @PathParam("username") String username) {
// 在线数-1
onlineClientNumber.decrementAndGet();
// 移除当前连接的session
onlineClientMap.remove(session.getId());
// 移除当前连接的user
onlineUserTimeMap.remove(session.getId());
log.error("onError---->Time:[{}],User:[{}],Session:[{}]，error:[{}]",
DateUtils.getTime(),
username,
session.getId(),
error.toString());
}
/**
* 客户端向服务端发送消息
*
* @param message 接受消息内容
* @param username 客户端用户名
*/
@OnMessage
public void onMsg(Session session, String message, @PathParam("username") String username) {
log.info("onMsg---->Time:[{}],User:[{}],Session[{}],message:[{}]",
DateUtils.getTime(),
username,
session.getId(),
message);
if (message != null && !message.equals("")) {
try {
JSONObject jsonMessage = JSONObject.parseObject(message);
if (jsonMessage.containsKey("messageType")
&& jsonMessage.get("messageType") != null) {
// 心跳消息
if (jsonMessage.get("messageType").equals("HeartBeat")) {
onlineUserTimeMap.put(session.getId(), System.currentTimeMillis());
log.info("onMsg---->HeartBeat:" + onlineUserTimeMap.toString());
}
}
} catch (Exception e) {
log.error("onMsg---->Exception:[{}]",e.toString());
}
}
}
/**
* 向所有客户端发送消息
*
* @param type 消息类别
* @param content 消息内容
*/
public static void sendMessageToAll(String type, String content) {
WebSocketMessage webSocketMessage = new WebSocketMessage();
webSocketMessage.setMessageId(DateUtils.getDateFormat(new Date(), DateUtils.FULL_TIME_PATTERN_ALL));
webSocketMessage.setMessageType(type);
webSocketMessage.setMessageContent(content);
webSocketMessage.setMessageTime(DateUtils.getTime());
String message = JSON.toJSON(webSocketMessage).toString();
// 获得Map的Key的集合
Set<String> sessionIdSet = onlineClientMap.keySet();
// 迭代Key集合
for (String sessionId : sessionIdSet) {
// 根据Key得到value
Session session = onlineClientMap.get(sessionId);
// 发送消息给客户端
session.getAsyncRemote().sendText(message);
}
}
/**
* 判断当前的Session是否超时
*/
public static void checkSession() {
if (onlineUserTimeMap != null && onlineUserTimeMap.size() > 0) {
// 获得Map的Key的集合
Set<String> sessionIdSet = onlineUserTimeMap.keySet();
// 获得当前时间戳
Long currentTime = System.currentTimeMillis();
// 迭代Key集合
for (String sessionId : sessionIdSet) {
// 根据Key得到value，即时间戳
Long time = onlineUserTimeMap.get(sessionId);
// 根据Key得到value，即Session
Session session = onlineClientMap.get(sessionId);
if (currentTime - time > timeInterval) {
// 此Session已超时
// 移除当前连接的session
onlineClientMap.remove(session.getId());
// 移除当前连接的user
onlineUserTimeMap.remove(session.getId());
log.info("checkSession---->remove:" + session.getId());
// 关闭session
try {
session.close();
} catch (Exception e) {
log.error("checkSession---->Exception:" + e.toString());
}
}
}
log.info("checkSession---->onlineUserTimeMap:" + onlineUserTimeMap.toString());
}
}
4. 定时任务
此模块主要用于心跳检测和对当前Session进行检测，将没有心跳返回的Session关闭。此处前端需要在收到心跳消息时，给服务器返回一条消息，证明前端还在。定时任务用的就是SpringBoot自带的定时任务模块。
@Component
@EnableScheduling
public class WebSocketTimer {
private static final Logger log = LoggerFactory.getLogger(WebSocket.class);
/**
* 心跳，每5分钟一次
*/
@Bean
@Scheduled(cron = "0 0/5 * * * ?")
public void WebSocketHeartBeat() {
log.info("WebSocketHeartBeat");
WebSocket.sendMessageToAll("HeartBeat", "");
}
/**
* 检测Session是否还在连接状态，每5分钟一次
* 在心跳发出的1分钟后执行
*/
@Bean
@Scheduled(cron = "0 1/5 * * * ?")
public void WebSocketCheckSession() {
log.info("WebSocketCheckSession");
WebSocket.checkSession();
}
}
5. 拦截修改
SpringBoot自带拦截器，将一些认为非法的请求过滤掉，如果你的项目里有SecurityConfig的配置，需要添加websocket地址。
httpSecurity
// 省略前面的配置代码
// 在此处添加
// websocket
.antMatchers("/websocket/**").anonymous()
// 省略后面的配置代码
6. 前端代码
前端就是连接，主要就是开启连接，接收消息，这里注意，前端接收到心跳消息，需要给服务器回一条心跳消息。这个前端代码就是测试代码，用户名使用的随机数序列，如果使用需要定用户名。
注：前端需要加个断开重连的逻辑，下面的代码里没有。
<script>
export default {
// name: "Line",
data() {
return {
username: "",
};
},
created() {
this.username = Math.random() + "";
this.initWebSocket();
},
mounted() {
window.addEventListener("message", function (e) {
});
},
methods: {
initWebSocket () {
const wsuri = 'ws://localhost:8088/websocket/' + this.username;
this.webSocketObject = new WebSocket(wsuri);
this.webSocketObject.onopen = this.webSocketOnOpen
this.webSocketObject.onmessage = this.webSocketOnMessage
this.webSocketObject.onerror = this.webSocketOnError
this.webSocketObject.onclose = this.webSocketOnClose
},
webSocketOnOpen(e){
console.log('与服务端连接打开->',e)
},
webSocketOnMessage(e){
console.log('来自服务端的消息->',e)
let tempData = JSON.parse(e.data)
if (tempData.messageType == "HeartBeat") {
const message = {
messageType: 'HeartBeat',
messageContent: ''
}
this.webSocketObject.send(JSON.stringify(message))
return
}
},
webSocketOnError(e){
console.log('与服务端连接异常->',e)
// 在此处加重连逻辑
},
webSocketOnClose(e){
console.log('与服务端连接关闭->',e)
// 在此处加重连逻辑
},
},
watch: {},
};
</script>
7. 其他配置
到这里在本地调试应该是没有问题了，但是在部署到服务器上，如果使用了nginx，可能在nginx还需要一些配置，包括添加websocket支持，这里需要注意的是，websocket连接超时的问题，如果不配置，nginx默认是一分钟没有消息发送，就会关闭，这里需要把参数改为10分钟(因为心跳消息是5分钟发送一次)。
具体配置如下：
在http节点下添加如下配置：
map $http_upgrade $connection_upgrade {
default upgrade;
'' close;
}
upstream websocket {
#ip_hash;
server localhost:8080;
}
在server节点下添加如下配置：其中最后一项就是超时时间的参数，这里设置为10分钟。
location /websocket {
proxy_pass http://localhost:8080;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 600s;
}
还有前端连接url，服务器IP的问题，上面前端代码写的是localhost，是在本地调试的时候用的。在服务器上部署之后，不能是localhost，要写成对应的服务器IP地址和端口号。
|