websocket实践

0 / 127

代码

/**
 * 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;
    }
}

经验

  1. 使用https的时候socket地址配置需要使用wss前缀,且nginx需要添加如下配置,否则会500 proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade";

这个时候http和https的登录其实不是一起的

  1. 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)
        }
    };


})();