springboot-websocket實現消息的主動推送

前段時間做項目，有一個場景是在后端接收到信號的時候將消息推送給前端，前端在展示給用戶。就是后端主動向前端推送消息的功能。這樣的場景可以讓前端輪詢，但是因為消息比較多，比較頻繁，所以如果讓前端輪詢的話，就有可能實時性不太好，而且由於http協議的請求頭比較長，進行輪詢會浪費大量的帶寬，服務器的壓力也會巨大。所以就采用的是websocket實現。

1.什么是websocket

WebSocket是一種在單個TCP連接上進行全雙工通信的協議。WebSocket使得客戶端和服務器之間的數據交換變得更加簡單，允許服務端主動向客戶端推送數據。在WebSocket API中，瀏覽器和服務器只需要完成一次握手，兩者之間就直接可以創建持久性的連接，並進行雙向數據傳輸。

2.原理

輪詢是采用固定的時間周期（比如1秒）請求后端接口，采用http協議進行通信，浪費資源。
而websocket是在首次建立連接的時候采用http協議，之后這個連接通道就一直存在着，直到自動關閉或者是手動關閉。當獲取 websocket 連接后，就可以通過 send() 方法來發送數據，並通過 onmessage 事件來接收數據。

3.代碼

首先引入pom依賴

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>

整個springboot項目的結構：

WebSocketConfig.java

package cn.fzkj.bootquick.config;

import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Component
public class WebSocketConfig {

    /**
     * 使用spring boot時，使用的是spring-boot的內置容器，
     * 如果要使用WebSocket，需要注入ServerEndpointExporter
     *
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

SocketUser.java

package cn.fzkj.bootquick.websocket;

import lombok.Data;

import javax.websocket.Session;

@Data
public class SocketUser {
    private Session session;
    private String uid;
}

這個類主要是用來封裝連接的信息。
接下來才是硬類：

package cn.fzkj.bootquick.websocket;


import cn.fzkj.bootquick.dto.APIResultCode;
import cn.fzkj.bootquick.dto.JsonDtoWrapper;
import cn.fzkj.bootquick.dto.MessageInfo;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.auth0.jwt.internal.org.apache.commons.lang3.StringUtils;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
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.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ServerEndpoint 注解是一個類層次的注解，它的功能主要是將目前的類定義成一個websocket服務器端,
 * 注解的值將被用於監聽用戶連接的終端訪問URL地址,客戶端可以通過這個URL來連接到WebSocket服務器端
 */
@Slf4j
@ServerEndpoint(value = "/mysocket/{userId}")
@Component
public class WebSocketService {
    //靜態變量，用來記錄當前在線連接數。應該把它設計成線程安全的。
    private static int onlineCount = 0;

    //concurrent包的線程安全Set，用來存放每個客戶端對應的MyWebSocket對象。
    // 若要實現服務端與單一客戶端通信的話，可以使用Map來存放，其中Key可以為用戶標識
    public static Map<String, SocketUser> userMap = new HashMap<>();
    private static Map<String, String> sessionUID = new ConcurrentHashMap<>();


    /**
     * 連接建立成功調用的方法
     *
     * @param session 可選的參數。session為與某個客戶端的連接會話，需要通過它來給客戶端發送數據
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        System.out.println("連接接收到的userId："+userId);
        SocketUser socketUser = new SocketUser();
        socketUser.setSession(session);
        socketUser.setUid(userId);
        userMap.put(socketUser.getUid(),socketUser);
        sessionUID.put(session.getId(), socketUser.getUid());
        onlineCount ++;
        log.info("有新連接加入！當前在線人數為" + getOnlineCount());
    }

    /**
     * 連接關閉調用的方法
     */
    @OnClose
    public synchronized void onClose(Session session) {
        String uid = sessionUID.get(session.getId());

        if (userMap.containsKey(uid)) {
            userMap.remove(uid);
            sessionUID.remove(session.getId());
            onlineCount --;
            log.info("success remove on connect=========================================>");
        }

        log.info("有一連接正常關閉！當前在線人數為" + uid + ":" + getOnlineCount());
    }

    /**
     * 收到客戶端消息后調用的方法
     *
     * @param message 客戶端發送過來的消息
     * @param session 可選的參數
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        if(StringUtils.isNotBlank(message) && message.charAt(0) == '{'){
            //解析發送的報文
            JSONObject jsonObject = JSONObject.parseObject(message);
            // 判斷接收到的消息
            String msg = jsonObject.getString("msg");
            System.out.println("從客戶端接收到的消息是："+msg);
            if ("heartbeat".equals(msg)){
                JsonDtoWrapper jo = new JsonDtoWrapper();
                jo.setCodeMsg(APIResultCode.HERAT_RESPONSE);
                try {
                    session.getBasicRemote().sendText(JSON.toJSONString(jo));
                } catch (Exception e) {
                    log.error("heart beat:", e);
                }
            }
            // 獲取接收者的id
            String recevierID = jsonObject.getString("recevierID");
            if (recevierID != null && !recevierID.equals("")){
                // 有接收者
                try{
                    SocketUser socketUser = userMap.get(recevierID);
                    if (null != socketUser){
                        socketUser.getSession().getBasicRemote().sendText(message);
                        log.info("success send");
                    }
                }catch (Exception e){
                    log.error("socket sendMessage", e);
                }
            }else {
                // 沒有接收者
                log.error("no recevier");
            }
        }else{
            try {
                session.getBasicRemote().sendText(message);
            } catch (Exception e) {
                log.error("heart beat:", e);
            }
        }
        log.info("來自客戶端的消息010:" + message);
    }

    /**
     * 發生錯誤時調用（這個有錯，文章后面會改正）
     *
     * @param session
     * @param error
     */
    @OnError
    public synchronized void onError(Session session, Throwable error) {
        log.info("發生錯誤");
        String uid = sessionUID.get(session.getId());

        if (userMap.containsKey(uid)) {
            userMap.remove(uid);
            sessionUID.remove(session.getId());
            onlineCount --;
            log.info("success remove on connect=========================================>");
        }

        log.info("有一連接錯誤關閉！當前在線人數為" + uid + ":" + getOnlineCount());

        error.printStackTrace();
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static JsonDtoWrapper sendMessage(MessageInfo messageInfo) {
        JsonDtoWrapper jo = new JsonDtoWrapper();
        try {
            SocketUser socketUser = userMap.get(messageInfo.getReceiveID());

            if (null != socketUser) {
                socketUser.getSession().getBasicRemote().sendText(new Gson().toJson(messageInfo));
                log.info("success send");
            } else {
                jo.setCodeMsg(APIResultCode.NOT_ONLINE);
                log.info("socket not online");
                return jo;
            }
            log.info("uid:" + messageInfo.getReceiveID());
            log.info("message:" + new Gson().toJson(messageInfo));
            jo.setCodeMsg(APIResultCode.SUCCESS);
            return jo;
        } catch (Exception e) {
            log.error("socket sendMessage", e);
            jo.setCodeMsg(APIResultCode.FAILURE);
            jo.setExtraMsg(e.toString());
            return jo;
        }
    }

}

onMessage方法為收到消息的時候調用的方法，具體業務自行擴展。

前端代碼：

<!DOCTYPE html>
<html>
<head>
    <title>Java后端WebSocket的Tomcat實現</title>
</head>
<body>
    Welcome<br/><input id="text" type="text"/>
    <button onclick="send()">發送消息</button>
    <hr/>
    <button onclick="closeWebSocket()">關閉WebSocket連接</button>
    <hr/>
    <div id="message"></div>
</body>

<script type="text/javascript">
    var websocket = null;
    //判斷當前瀏覽器是否支持WebSocket
    if ('WebSocket' in window) {
        websocket = new WebSocket("ws://localhost:8081/mysocket/10");
    }
    else {
        alert('當前瀏覽器 Not support websocket')
    }

    //連接發生錯誤的回調方法
    websocket.onerror = function () {
        setMessageInnerHTML("WebSocket連接發生錯誤");
    };

    //連接成功建立的回調方法
    websocket.onopen = function () {
        setMessageInnerHTML("WebSocket連接成功");
    }

    //接收到消息的回調方法
    websocket.onmessage = function (event) {
		console.log(event.data);
		var msg = JSON.parse(event.data);
        setMessageInnerHTML(msg['msg']+"&#13;");
    }

    //連接關閉的回調方法
    websocket.onclose = function () {
        setMessageInnerHTML("WebSocket連接關閉");
    }

    //監聽窗口關閉事件，當窗口關閉時，主動去關閉websocket連接，防止連接還沒斷開就關閉窗口，server端會拋異常。
    window.onbeforeunload = function () {
        closeWebSocket();
    }

    //將消息顯示在網頁上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }

    //關閉WebSocket連接
    function closeWebSocket() {
        websocket.close();
    }

    //發送消息
    function send() {
        var message = document.getElementById('text').value;
        websocket.send('{"recevierID":"11","msg":"'+message+'"}');
		setMessageInnerHTML(message+"&#13;");
    }
</script>
</html>

效果展示：

關於我使用到的工具類：
JsonDtoWrapper.java

package cn.fzkj.bootquick.dto;

import lombok.Data;

@Data
public class JsonDtoWrapper {
    private String code;
    private String msg;
    private String extraMsg;

    public void setCodeMsg(APIResultCode co) {
        this.code = co.getCode();
        this.msg = co.getMsg();
        this.extraMsg = co.getExtraMsg();
    }
}

---

持續更新~~

---

（2020/7/12）
在項目中使用上述代碼，遇到了一個問題，同一個clientID被斷掉兩次，所以導致第二次會有異常（NullPointException）。最終呈現的效果就是，頁面沒反應。。這個問題着實困擾了我好幾天，最開始以為是代碼同步的問題，加上鎖之后問題確實在當時被解決了。但是過了幾天又出現了。。在網上找各種的方案，，最終經過好幾天，這個問題才終於被徹底的解決了。
原因是：OnError發生時，會觸發OnClose方法。所以在之前的代碼中，這兩個方法中都對clientID和session進行了處理，所以在OnClose中，會報NullPointException異常。
解決方案就是：在OnError中不能處理任何與clinetID和session有關的邏輯。。
上面OnError方法修改如下：

    /**
     * 進入error方法后，會觸發close方法，所以error方法中不能處理和uid、sessionid有關的操作，不然close方法中還會處理。會報空指針異常
     *
     * @param session
     * @param error
     */
    @OnError
    public synchronized void error(Session session, Throwable error) {
        log.error("OnError====>" + error.getMessage());
    }

---

持續更新~