一、概述
1.1 简介
WebSocket是HTML5下一种新的协议（websocket协议本质上是一个基于tcp的协议），它实现了浏览器与服务器全双工通信，能更好的节省服务器资源和带宽并达到实时通讯的目的，Websocket是一个持久化的协议。
1.2 原理
- websocket约定了一个通信的规范，通过一个握手的机制，客户端和服务器之间能建立一个类似tcp的连接，从而方便它们之间的通信
- 在websocket出现之前，web交互一般是基于http协议的短连接或者长连接
- websocket是一种全新的协议，不属于http无状态协议，协议名为
ws- 优点：减少资源消耗;实时推送不用等待客户端的请求;减少通信量;
- 缺点：少部分浏览器不支持，不同浏览器支持的程度和方式都不同。
- 应用场景
- 智慧大屏
- 消息提醒通知
- …
二、具体实现
2.1 编写pom
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
2.2 websocket配置
@Component
public class WebsocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter(){
return new ServerEndpointExporter();
}
}
2.3 websocket服务端
@ServerEndpoint("/wbsocket")
@Component
@Slf4j
public class Websocket {
// 记录连接的客户端
public static Map<String, Session> clients = new ConcurrentHashMap<>();
/**
* userId关联sid（解决同一用户id，在多个web端连接的问题）
*/
public static Map<String, Set<String>> cons = new ConcurrentHashMap<>();
private String sid = null;
// 一些记录发送消息状态
private static int initFlag = 0;
private static int tempFlag = 0;
// 区分新旧消息的变量
private static int sum = 0;
/**
* 连接成功后调用的方法
*
* @param session 会话
*/
@OnOpen
public void onOpen(Session session) {
this.sid = UUID.randomUUID().toString();
clients.put(this.sid, session);
log.info(this.sid + "连接开启！");
}
/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose() {
log.info(this.sid + "连接断开！");
clients.remove(this.sid);
}
/**
* 判断是否连接的方法
*
* @return 连接true, 未连接false
*/
public static boolean isServerClose() {
if (CollectionUtils.isEmpty(Websocket.clients.values())) {
log.info("已断开");
return true;
} else {
log.info("已连接");
return false;
}
}
/**
* 发送给所有用户
*
* @param noticeType 提醒类型
*/
public static boolean sendMessage(String noticeType, int count) {
if (sum != count) {
WsResp noticeWsResp = new WsResp();
noticeWsResp.setNoticeType(noticeType);
sendMessage(noticeWsResp);
sum = count;
}
//判断前端是否 回复了 收到消息 相等没收到，不相等 收到
if (initFlag == tempFlag) {
WsResp noticeWsResp = new WsResp();
noticeWsResp.setNoticeType(noticeType);
sendMessage(noticeWsResp);
} else {
tempFlag = initFlag;
log.info("收到消息了，别发同一个消息了");
return true;
}
tempFlag = initFlag;
log.info("没收到消息继续发");
return false;
}
/**
* 发送给所有用户
*
* @param websocketResp websocket的返回
*/
public static void sendMessage(WsResp websocketResp) {
String message = JSONObject.toJSONString(websocketResp);
for (Session session1 : Websocket.clients.values()) {
try {
session1.getBasicRemote().sendText(message);
} catch (IOException e) {
log.error("给所有用户的消息发送失败, msg: {}", message, e);
}
}
}
/**
* 根据用户id发送给某一个用户
**/
public static void sendMessageByUserId(String userId, WsResp wsResp) {
if (!StringUtils.isBlank(userId)) {
String message = JSONObject.toJSONString(wsResp);
Set<String> clientSet = cons.get(userId);
if (clientSet != null) {
for (String sid : clientSet) {
Session session = clients.get(sid);
if (session != null) {
try {
session.getBasicRemote().sendText(message);
} catch (IOException e) {
log.error("用户消息发送失败, sid: {}, msg: {}", sid, message, e);
}
}
}
}
}
}
/**
* 收到客户端消息后调用的方法
*
* @param message 消息
* @param session 会话
*/
@OnMessage
public void onMessage(String message, Session session) {
log.info("收到来自窗口" + "的信息:" + message);
if ("已接收到消息".equals(message)) {
//收到消息，改变flag的值
log.info("前端已经收到消息，开始改变 initFlag的值，此时initFlag= " + initFlag);
initFlag = initFlag + 1;
log.info("前端已经收到消息，已经改变 initFlag的值，此时initFlag== " + initFlag);
}
}
/**
* 发生错误时的回调函数
*
* @param error 错误
*/
@OnError
public void onError(Throwable error) {
log.info("错误");
}
}
2.4 service & impl 编写
public interface RecycleCustomerService {
void sendMsg(WsMsgVO msgVO);
}
Slf4j
@Service
public class RecycleCustomerServiceImpl implements RecycleCustomerService {
private static final AtomicInteger count = new AtomicInteger(0);
@Override
public void sendMsg(WsMsgVO msgVO) {
// 测试websocket,实现新增客户往前端推送消息，直到前端回复
boolean flag = false;
do {
// 休息300毫秒
try { Thread.sleep(300); } catch (InterruptedException e) { log.error("休息时出错~~~~~~~", e);}
// 往前端发送消息
flag = Websocket.sendMessage("实时数据: " + msgVO.toString(), count.get());
if (flag) {
log.info("停止往前端发送数据, 因为 resultFlag 为: {}, 说明前端接收到消息", flag);
} else {
log.info("往前端发送数据, 因为 resultFlag 为: {}", flag);
}
} while (!flag);
log.info("停止往前端发送数据, 因为 delFlag 为: {}", flag);
int count = RecycleCustomerServiceImpl.count.incrementAndGet();
log.info("当前count: {}", count);
}
}
2.5 index.html编写
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>SseEmitter</title>
</head>
<body>
<div id="message"></div>
</body>
<script>
let limitConnect = 0;
init();
function init() {
// 开启 wbsocket 服务的ip地址 ws:// + ip地址 + 访问路径
let ws = new WebSocket('ws://127.0.0.1:8888/wbsocket');
// 获取连接状态
console.log('ws连接状态：' + ws.readyState);
// 监听是否连接成功
ws.onopen = function () {
console.log('ws连接状态：' + ws.readyState);
limitConnect = 0;
//连接成功则发送一个数据
ws.send('我们建立连接啦');
}
// 接听服务器发回的信息并处理展示
ws.onmessage = function (data) {
console.log('接收到来自服务器的消息：');
console.log(data);
//接收到 消息后给后端发送的 确认收到消息，后端接收到后 不再重复发消息
ws.send('已接收到消息');
//完成通信后关闭WebSocket连接
// ws.close();
}
// 监听连接关闭事件
ws.onclose = function () {
// 监听整个过程中websocket的状态
console.log('ws连接状态：' + ws.readyState);
reconnect();
}
// 监听并处理error事件
ws.onerror = function (error) {
console.log(error);
}
}
function reconnect() {
limitConnect++;
console.log("重连第" + limitConnect + "次");
setTimeout(function () {
init();
}, 2000);
}
</script>
</html>
三、运行并测试
调用: http://127.0.0.1:8888/ws/test
可以看到控制台打印
websocket测试成功！
代码地址: websocket-demo
