springboot整合SocketIO實時通訊

阿新 • • 發佈：2018-12-16

概述

基於 socket.io 來說，採用 node 實現更加合適，本文使用兩個後端的Java開源框架實現。

業務需求是將之前通過輪詢方式調動RESTFul API改成使用WebSocket長連線方式，實現要伺服器實時的推送訊息，另外還要實時監控POS機的線上狀態等。

引入依賴

<!-- netty-socketio-->
<dependency>
    <groupId>com.corundumstudio.socketio</groupId>
    <artifactId>netty-socketio</artifactId>
    <version>1.7.13</version>
</dependency>
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-resolver</artifactId>
    <version>4.1.15.Final</version>
</dependency>
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-transport</artifactId>
    <version>4.1.15.Final</version>
</dependency>
<dependency>
    <groupId>io.socket</groupId>
    <artifactId>socket.io-client</artifactId>
    <version>1.0.0</version>
</dependency>

先來服務端程式爽一把，話不多說，先上程式碼：

伺服器程式碼

public class NamespaceSocketServer {
    private static final Logger logger = LoggerFactory.getLogger(NamespaceSocketServer.class);

    public static void main(String[] args) {
        /*
         * 建立Socket，並設定監聽埠
         */
        Configuration config = new Configuration();
        //設定主機名
        config.setHostname("localhost");
        //設定監聽埠
        config.setPort(9092);
        // 協議升級超時時間（毫秒），預設10秒。HTTP握手升級為ws協議超時時間
        config.setUpgradeTimeout(10000);
        // Ping訊息超時時間（毫秒），預設60秒，這個時間間隔內沒有接收到心跳訊息就會發送超時事件
        config.setPingTimeout(180000);
        // Ping訊息間隔（毫秒），預設25秒。客戶端向伺服器傳送一條心跳訊息間隔
        config.setPingInterval(60000);
        // 連線認證，這裡使用token更合適
        config.setAuthorizationListener(new AuthorizationListener() {
            @Override
            public boolean isAuthorized(HandshakeData data) {
                // String token = data.getSingleUrlParam("token");
                // String username = JWTUtil.getSocketUsername(token);
                // return JWTUtil.verifySocket(token, "secret");
                return true;
            }
        });

        final SocketIOServer server = new SocketIOServer(config);

        /*
         * 新增連線監聽事件，監聽是否與客戶端連線到伺服器
         */
        server.addConnectListener(new ConnectListener() {
            @Override
            public void onConnect(SocketIOClient client) {
                // 判斷是否有客戶端連線
                if (client != null) {
                    logger.info("連線成功。clientId=" + client.getSessionId().toString());
                    client.joinRoom(client.getHandshakeData().getSingleUrlParam("appid"));
                } else {
                    logger.error("並沒有人連線上。。。");
                }
            }
        });

        /*
         * 新增監聽事件，監聽客戶端的事件
         * 1.第一個引數eventName需要與客戶端的事件要一致
         * 2.第二個引數eventClase是傳輸的資料型別
         * 3.第三個引數listener是用於接收客戶端傳的資料，資料型別需要與eventClass一致
         */
        server.addEventListener("login", LoginRequest.class, new DataListener<LoginRequest>() {
            @Override
            public void onData(SocketIOClient client, LoginRequest data, AckRequest ackRequest) {
                logger.info("接收到客戶端login訊息：code = " + data.getCode() + ",body = " + data.getBody());
                // check is ack requested by client, but it's not required check
                if (ackRequest.isAckRequested()) {
                    // send ack response with data to client
                    ackRequest.sendAckData("已成功收到客戶端登入請求", "yeah");
                }
                // 向客戶端傳送訊息
                List<String> list = new ArrayList<>();
                list.add("登入成功，sessionId=" + client.getSessionId());
                // 第一個引數必須與eventName一致，第二個引數data必須與eventClass一致
                String room = client.getHandshakeData().getSingleUrlParam("appid");
                server.getRoomOperations(room).sendEvent("login", list.toString());
            }
        });
        //啟動服務
        server.start();
    }
}

老規矩，先上程式碼爽爽

Android客戶端

public class SocketClient {
    private static Socket socket;
    private static final Logger logger = LoggerFactory.getLogger(SocketClient.class);

    public static void main(String[] args) throws URISyntaxException {
        IO.Options options = new IO.Options();
        options.transports = new String[]{"websocket"};
        options.reconnectionAttempts = 2;     // 重連嘗試次數
        options.reconnectionDelay = 1000;     // 失敗重連的時間間隔(ms)
        options.timeout = 20000;              // 連線超時時間(ms)
        options.forceNew = true;
        options.query = "username=test1&password=test1&appid=com.xncoding.apay2";
        socket = IO.socket("http://localhost:9092/", options);
        socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                // 客戶端一旦連線成功，開始發起登入請求
                LoginRequest message = new LoginRequest(12, "這是客戶端訊息體");
                socket.emit("login", JsonConverter.objectToJSONObject(message), (Ack) args1 -> {
                    logger.info("回執訊息=" + Arrays.stream(args1).map(Object::toString).collect(Collectors.joining(",")));
                });
            }
        }).on("login", new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                logger.info("接受到伺服器房間廣播的登入訊息：" + Arrays.toString(args));
            }
        }).on(Socket.EVENT_CONNECT_ERROR, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                logger.info("Socket.EVENT_CONNECT_ERROR");
                socket.disconnect();
            }
        }).on(Socket.EVENT_CONNECT_TIMEOUT, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                logger.info("Socket.EVENT_CONNECT_TIMEOUT");
                socket.disconnect();
            }
        }).on(Socket.EVENT_PING, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                logger.info("Socket.EVENT_PING");
            }
        }).on(Socket.EVENT_PONG, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                logger.info("Socket.EVENT_PONG");
            }
        }).on(Socket.EVENT_MESSAGE, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                logger.info("-----------接受到訊息啦--------" + Arrays.toString(args));
            }
        }).on(Socket.EVENT_DISCONNECT, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                logger.info("客戶端斷開連線啦。。。");
                socket.disconnect();
            }
        });
        socket.connect();
    }
}

關於心跳機制

根據 Socket.IO文件解釋，客戶端會定期傳送心跳包，並觸發一個ping事件和一個pong事件，如下：

ping Fired when a ping packet is written out to the server.
pong Fired when a pong is received from the server. Parameters:
- Number number of ms elapsed since ping packet (i.e.: latency)

這裡最重要的兩個伺服器引數如下：

pingTimeout (Number): how many ms without a pong packet to consider the connection closed (60000)
pingInterval (Number): how many ms before sending a new ping packet (25000).

也就是說握手協議的時候，客戶端從伺服器拿到這兩個引數，一個是ping訊息的傳送間隔時間，一個是從伺服器返回pong訊息的超時時間，客戶端會在超時後斷開連線。心跳包傳送方向是客戶端向伺服器端傳送，以維持線上狀態。

關於斷線和超時

關閉瀏覽器、直接關閉客戶端程式、kill程序、主動執行disconnect方法都會導致立刻產生斷線事件。而客戶端把網路斷開，伺服器端在 pingTimeout ms後產生斷線事件、客戶端在 pingTimeout ms後也產生斷線事件。

實際上，超時後會產生一個斷線事件，叫”disconnect”。客戶端和伺服器端都可以對這個事件作出應答，釋放連線。

客戶端程式碼：

.on(Socket.EVENT_DISCONNECT, new Emitter.Listener() {
    @Override
    public void call(Object... args) {
        logger.info("客戶端斷開連線啦。。。");
        socket.disconnect();
    }
});

下面是客戶端日誌：連上伺服器後，斷開網路。超過了心跳超時時間後，產生斷線事件。如果客戶端不主動斷開連線的話，會自動重連，這時候發現連線不上，又產生連線錯誤事件，然後重試2次，都失敗後自動斷開連線了。

SocketClient - 回執訊息=伺服器已成功收到客戶端登入請求,yeah
SocketClient - Socket.EVENT_PING
SocketClient - Socket.EVENT_PONG
SocketClient - 客戶端斷開連線啦。。。
SocketClient - Socket.EVENT_CONNECT_ERROR

伺服器端程式碼：

server.addDisconnectListener(new DisconnectListener() {
    @Override
    public void onDisconnect(SocketIOClient client) {
        System.out.println("伺服器收到斷線通知... sessionId=" + client.getSessionId());
    }
});

伺服器邏輯是，如果在心跳超時後，就直接斷開這個連線，並且產生一個斷開連線事件。

伺服器通過netty處理心跳包ping/pong的日誌如下：

1
2
3

WebSocket08FrameDecoder - Decoding WebSocket Frame opCode=1
WebSocket08FrameDecoder - Decoding WebSocket Frame length=1
WebSocket08FrameEncoder - Encoding WebSocket Frame opCode=1 length=1

瀏覽器客戶端演示

對於netty-socketio有一個demo工程，裡面通過一個網頁的聊天小程式演示了各種使用方法。