最近有点小时间，上个项目正好用到了websocket实现广播消息来着，现在来整理一下之前的一些代码，分享给大家。
WebSocket协议是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通信——允许服务器主动发送信息给客户端。
开发工具：IntelliJ IDEA
运行环境：SpringBoot2.x、ReconnectingWebSocket、JDK1.8+、Maven 3.6 +
ReconnectingWebSocket 是一个小型的 JavaScript 库，封装了 WebSocket API 提供了在连接断开时自动重连的机制。
只需要简单的将：
ws = new WebSocket('ws://....');替换成：
ws = new ReconnectingWebSocket('ws://....');WebSocket 属性ws.readyState：
0 - 表示连接尚未建立。
1 - 表示连接已建立，可以进行通信。
2 - 表示连接正在进行关闭。
3 - 表示连接已经关闭或者连接不能打开。
WebSocket事件：
事件 事件处理程序 描述 open ws.onopen 连接建立时触发 message ws.onmessage 客户端接收服务端数据时触发 error ws.onerror 通信发生错误时触发 close ws.onclose 连接关闭时触发 WebSocket方法：
方法 描述 Socket.send() 使用连接发送数据 Socket.close() 关闭连接
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<!-- springbooot 集成 websocket -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.71</version>
</dependency>
</dependencies>
index.html
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8"/>
<title>SpringBoot-ws</title>
<script src="../js/reconnectingwebsocket.js" type="text/javascript" charset="utf-8"></script>
<!-- <script src="../js/sockjs.min.js" type="text/javascript" charset="utf-8"></script>-->
<script src="../js/jquery.min.js" type="text/javascript" charset="utf-8"></script>
<link rel="stylesheet" type="text/css" href="../css/style.css">
</head>
<body>
<div id="info">
<div>发送人：<input type="text" id="suer" required="required" placeholder="请输入发送人"></div>
<div>接收人：<input type="text" id="ruser" required="required" placeholder="请输入接收人"></div>
</div>
<div id="index">
</div>
<div class="msg">
<textarea id="send_content" placeholder="在此输入消息..."></textarea>
</div>
<div class="ibtn c">
<button onclick=openWebsocket()>开启连接</button>
<button onclick=closeWebsocket()>关闭连接</button>
<button onclick=sendMessage()>发送消息</button>
</div>
<script type="text/javascript">
document.getElementById('send_content').focus();
var websocket = null;
//关闭websocket
function closeWebsocket() {
//3代表已经关闭
if (3 != websocket.readyState) {
websocket.close();
} else {
alert("websocket之前已经关闭");
}
}
// 开启websocket
function openWebsocket() {
username = $("#suer").val()
if (username != "") {
//当前浏览前是否支持websocket
if ("WebSocket" in window) {
websocket = new ReconnectingWebSocket("ws://localhost:8080/send/" + username);
websocket.reconnectInterval = 3000 //每3s进行一次重连，默认是每秒
} else if ('MozWebSocket' in window) {
websocket = new MozWebSocket("ws://localhost:8080/send/" + username);
} else {
//低版本
websocket = new SockJS("http://localhost:8080/sockjs/send/" + username);
}
}
websocket.onopen = function (event) {
setMessage("打开连接");
}
websocket.onclose = function (event) {
setMessage("关闭连接");
}
websocket.onmessage = function (event) {
// setMessage(event.data);
setMessageTxt(event.data)
}
websocket.onerror = function (event) {
setMessage("连接异常，正在重连中...");
}
//监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
window.onbeforeunload = function () {
closeWebsocket();
}
}
//将消息显示在网页上
function setMessage(message) {
alert(message)
}
function setMessageTxt(message) {
mObj = JSON.parse(message)
var div = document.createElement('div')
div.innerHTML = "<div class='name l'><h2>" + mObj['from_topic'] + "</h2></div>" +
"<div class='content w l'>" + mObj['content'] + "</div>"
div.setAttribute("class", "from_info")
document.getElementById('index').appendChild(div)
}
// 发送消息
function sendMessage() {
//1代表正在连接
if (1 == websocket.readyState) {
var message = document.getElementById('send_content').value;
var div = document.createElement('div')
div.innerHTML = "<div class='name r rcontent'><h2> Me </h2></div>" +
"<div class='content w r'>" + message + "</div>"
div.setAttribute("class", "send_info")
document.getElementById('index').appendChild(div)
ruser = document.getElementById("ruser").value;
message = "{'content':'" + message + "','to_topic':'" + ruser + "'}"
websocket.send(message);
} else {
alert("websocket未连接");
}
document.getElementById('send_content').value = "";
document.getElementById('send_content').focus();
}
</script>
</body>
</html>
package com.jhzhong.swchat.controller;
import com.jhzhong.swchat.websocket.WebSocketServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
@Controller
public class SWController {
@Autowired
private WebSocketServer webSocketServer;
/**
* author: jhzhong95@gmail.com
* date: 2020-06-24 12:35 AM
* desc: 跳转index.html页面
* @return
*/
@RequestMapping("/")
public String index() {
return "index";
}
}
WebSocketConfig.java 类，开启WebSocket支持。 package com.jhzhong.swchat.websocket;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* author: jhzhong95@gmail.com
* date: 2020-06-24 12:28 AM
* desc: 开启WebSocket支持
*/
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter(){
return new ServerEndpointExporter();
}
}
WebSocketServer.java
package com.jhzhong.swchat.websocket;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import freemarker.log.Logger;
import org.apache.commons.lang.StringUtils;
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.concurrent.ConcurrentHashMap;
/**
* author: jhzhong95@gmail.com
* date: 2020-06-24 12:40 AM
* desc: WebSocket服务端
*/
@ServerEndpoint("/send/{topic}")
@Component
public class WebSocketServer {
static Logger logger = Logger.getLogger("WebSocketServer");
/**
* 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
*/
private static int onlineCount = 0;
/**
* concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
*/
private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
/**
* 与某个客户端的连接会话，需要通过它来给客户端发送数据
*/
private Session session;
/**
* 接收频道topic
*/
private String topic = "";
/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("topic") String topic) {
this.session = session;
this.topic = topic;
if (webSocketMap.containsKey(topic)) {
webSocketMap.remove(topic);
webSocketMap.put(topic, this);
//加入set中
} else {
webSocketMap.put(topic, this);
//加入set中
addOnlineCount();
//在线数加1
}
logger.info("用户连接:" + topic + ",当前在线人数为:" + getOnlineCount());
try {
sendMessage("连接成功");
} catch (IOException e) {
logger.error("用户:" + topic + ",网络异常!!!!!!");
}
}
/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose() {
if (webSocketMap.containsKey(topic)) {
webSocketMap.remove(topic);
//从set中删除
subOnlineCount();
}
logger.info("用户退出:" + topic + ",当前在线人数为:" + getOnlineCount());
}
/**
* 收到客户端消息后调用的方法
*
* @param message 客户端发送过来的消息
*/
@OnMessage
public void onMessage(String message, Session session) {
logger.info("用户:" + topic + ",信息:" + message);
//可以群发消息
//消息保存到数据库、redis
if (StringUtils.isNotBlank(message)) {
try {
//解析发送的报文
JSONObject jsonObject = JSON.parseObject(message);
//追加发送人(防止串改)
jsonObject.put("from_topic", this.topic);
String to_topic = jsonObject.getString("to_topic");
//传送给对应toUserId用户的websocket
if (StringUtils.isNotBlank(to_topic) && webSocketMap.containsKey(to_topic)) {
webSocketMap.get(to_topic).sendMessage(jsonObject.toJSONString());
} else {
logger.error("请求的to_topic:" + to_topic + "不在该服务器上");
//否则不在这个服务器上，发送到mysql或者redis
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
/**
* @param session
* @param error
*/
@OnError
public void onError(Session session, Throwable error) {
logger.error("用户错误:" + this.topic + ",原因:" + error.getMessage());
error.printStackTrace();
}
/**
* 实现服务器主动推送
*/
public void sendMessage(String message) throws IOException {
this.session.getBasicRemote().sendText(message);
}
/**
* 发送自定义消息
*/
public static void sendInfo(String message, @PathParam("topic") String topic) throws IOException {
logger.info("发送消息到:" + topic + "，信息:" + message);
if (StringUtils.isNotBlank(topic) && webSocketMap.containsKey(topic)) {
webSocketMap.get(topic).sendMessage(message);
} else {
logger.error("用户" + topic + ",不在线！");
}
}
public static synchronized int getOnlineCount() {
return onlineCount;
}
public static synchronized void addOnlineCount() {
WebSocketServer.onlineCount++;
}
public static synchronized void subOnlineCount() {
WebSocketServer.onlineCount--;
}
}
首页截图
视频效果
SpringBoot+WebSocket实现消息广播
如需源码请参考： sw-chat.zip 源码下载