代码
/**
* WebScoket配置处理器
*/
@Configuration
public class WebSocketConfig {
/**
* ServerEndpointExporter 作用
* 这个Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint
*
* @return
*/
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
@ServerEndpoint("/ws/{loginName}")
@Component
@JwtIgnore
public class WebSocketServer {
private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class);
//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
private static AtomicInteger onlineNum = new AtomicInteger();
//concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
private static ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();
//发送消息
private static void sendMessage(Session session, Message message) throws IOException {
String msgStr = JSON.toJSONString(message, true);
if (session != null) {
synchronized (session) {
logger.info("发送数据：" + msgStr);
session.getBasicRemote().sendText(msgStr);
}
}
}
//给指定用户发送信息
public static void sendInfo(String userName, Message message) {
Session session = sessionPools.get(userName);
try {
sendMessage(session, message);
} catch (Exception e) {
e.printStackTrace();
}
}
// 群发消息
public static void broadcast(Message message) {
for (Session session : sessionPools.values()) {
try {
sendMessage(session, message);
} catch (Exception e) {
e.printStackTrace();
}
}
}
//建立连接成功调用
@OnOpen
public void onOpen(Session session, @PathParam(value = "loginName") String userName) {
sessionPools.put(userName, session);
addOnlineCount();
logger.info(userName + "已登录！当前人数为" + onlineNum);
}
//关闭连接时调用
@OnClose
public void onClose(@PathParam(value = "loginName") String userName) {
sessionPools.remove(userName);
subOnlineCount();
logger.info(userName + "关闭下载页面！当前人数为" + onlineNum);
}
//收到客户端信息后，根据接收人的username把消息推下去或者群发
// to=-1群发消息
@OnMessage
public void onMessage(String message) throws IOException {
System.out.println("server get" + message);
Message msg = JSON.parseObject(message, Message.class);
msg.setDate(new Date());
if (msg.getTo().equals("-1")) {
broadcast(msg);
} else {
sendInfo(msg.getTo(), msg);
}
}
//错误时调用
@OnError
public void onError(Session session, Throwable throwable) {
logger.error("发生错误");
throwable.printStackTrace();
}
private static void addOnlineCount() {
onlineNum.incrementAndGet();
}
private static void subOnlineCount() {
onlineNum.decrementAndGet();
}
public static AtomicInteger getOnlineNumber() {
return onlineNum;
}
public static ConcurrentHashMap<String, Session> getSessionPools() {
return sessionPools;
}
}
经验
- 使用https的时候socket地址配置需要使用wss前缀，且nginx需要添加如下配置，否则会500
proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade";
这个时候http和https的登录其实不是一起的
- nginx连接超时时间导致出错
java.io.EOFException at org.apache.tomcat.util.net.NioEndpoint$NioSocketWrapper.fillReadBuffer(NioEndpoint.java:1208) at org.apache.tomcat.util.net.NioEndpoint$NioSocketWrapper.read(NioEndpoint.java:1142) at org.apache.tomcat.websocket.server.WsFrameServer.onDataAvailable(WsFrameServer.java:72) at org.apache.tomcat.websocket.server.WsFrameServer.doOnDataAvailable(WsFrameServer.java:171) at org.apache.tomcat.websocket.server.WsFrameServer.notifyDataAvailable(WsFrameServer.java:151) at org.apache.tomcat.websocket.server.WsHttpUpgradeHandler.upgradeDispatch(WsHttpUpgradeHandler.java:148) at org.apache.coyote.http11.upgrade.UpgradeProcessorInternal.dispatch(UpgradeProcessorInternal.java:54) at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:53) at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:791) at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1417) at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) at java.lang.Thread.run(Thread.java:745)
nginx默认超时时间是60s，超过时间连接就断了，后修改了keepalive_timeout：
keepalive_timeout 300;
后来查原来使用websocket最好加上心跳检测，所以就加上了：
(function (undefined) {
var CHAT = {
socketUrl: "ws://goshare.site/main/ws",
initChatSocket: function (userInfo) {
return initChatSocket(userInfo);
}
};
window.CHAT = window.CHAT || CHAT;
var websocket;
var userDetailInfo;
function initChatSocket(userInfo) {
userDetailInfo = userInfo;
var userLoginName;
// 指定websocket路径
//获取当前登录状态
userLoginName = userInfo.loginName;
if ('WebSocket' in window) {
websocket = new WebSocket(CHAT.socketUrl + "/" + userLoginName);
}
init();
}
function init() {
//连接成功建立的回调方法
websocket.onopen = function (event) {
console.log("WebSocket:已连接");
//心跳检测重置
heartCheck.reset().start();
};
//接收到消息的回调方法
websocket.onmessage = function (event) {
var data = JSON.parse(event.data);
// 普通消息
// 接收服务端的实时消息并添加到HTML页面中
layer.confirm("来自" + data.from + "的消息：" + data.text, {icon: 6});
//拿到任何消息 心跳检测重置
heartCheck.reset().start();
};
//连接发生错误的回调方法
websocket.onerror = function (event) {
console.log("WebSocket:发生错误");
reconnect();
};
//连接关闭的回调方法
websocket.onclose = function (event) {
console.log("WebSocket:已关闭");
heartCheck.reset();//心跳检测
reconnect();
};
//监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
window.onbeforeunload = function () {
websocket.close();
};
//关闭连接
function closeWebSocket() {
websocket.close();
}
//发送消息
function send(message) {
websocket.send(message);
}
}
//避免重复连接
var lockReconnect = false, tt;
/**
* websocket重连
*/
function reconnect() {
if (lockReconnect) {
return;
}
lockReconnect = true;
tt && clearTimeout(tt);
tt = setTimeout(function () {
console.log('重连中...');
lockReconnect = false;
initChatSocket(userDetailInfo);
}, 4000);
}
/**
* websocket心跳检测
*/
var heartCheck = {
timeout: 5000,
timeoutObj: null,
serverTimeoutObj: null,
reset: function () {
clearTimeout(this.timeoutObj);
clearTimeout(this.serverTimeoutObj);
return this;
},
start: function () {
var self = this;
this.timeoutObj && clearTimeout(this.timeoutObj);
this.serverTimeoutObj && clearTimeout(this.serverTimeoutObj);
this.timeoutObj = setTimeout(function () {
//这里发送一个心跳，后端收到后，返回一个心跳消息，
//onmessage拿到返回的心跳就说明连接正常
var data = {};
data["text"] = 'ping';
websocket.send(JSON.stringify(data));
console.log('ping');
self.serverTimeoutObj = setTimeout(function () { // 如果超过一定时间还没重置，说明后端主动断开了
console.log('关闭服务');
websocket.close();
//如果onclose会执行reconnect，我们执行 websocket.close()就行了.如果直接执行 reconnect 会触发onclose导致重连两次
}, self.timeout)
}, this.timeout)
}
};
})();