springboot使用websocket進行實時傳送信息實戰(包含服務端和客戶端)

1.背景

---

 

公司對接告警信息平台，需要做一個服務，作為websocket客戶端去接收傳過來的信息，實時返回信息並對信息進行處理入庫。

 

2.實現方案

---

 

本來想用一個服務，對信息進行接收和處理。但是基於之前的經驗，為了防止服務部署重啟的時候丟失信息，改用兩個服務：1.collcet接收服務，2.deal-send處理入庫服務。collect服務作為websocket客戶端，進行接收傳過來的信息，並同時作為websocket服務端，傳信息給deal-send服務；deal-send作為websocket客戶端，接收collect傳過來的信息，進行處理入庫。

 

3.代碼實例

---

 

• collect服務主要依賴

pom.xml

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

        <!--websocket作為客戶端-->
        <dependency>
            <groupId>org.java-websocket</groupId>
            <artifactId>Java-WebSocket</artifactId>
            <version>1.4.0</version>
        </dependency>

        <!--使用Spring RetryTemplate進行重試-->
        <dependency>
            <groupId>org.springframework.retry</groupId>
            <artifactId>spring-retry</artifactId>
            <version>1.2.4.RELEASE</version>
        </dependency>

 

• deal-send服務主要依賴

pom.xml

        <!--websocket作為客戶端-->
        <dependency>
            <groupId>org.java-websocket</groupId>
            <artifactId>Java-WebSocket</artifactId>
            <version>1.4.0</version>
        </dependency>

 

•  collect配置類

WebSocketConfig.java

package com.newland.auto.pilot.collect.config;

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

/**
 * @author:wk
 * @date:2020/6/10 開啟WebSocket支持
 */

@Configuration
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

}

 

• collect服務websocket客戶端，由於告警信息平台偶爾會斷網，websocket的長連接就會被斷開，所以在onClose方法里加入了spring-retry提供的重試方法，在斷開websocket連接的時候會進行重連操作。onMessage方法里加了發送信息到deal-send服務的調用，在接收到告警信息平台的時候，會將信息發到deal-send服務。

MyWebSocketCient.java

package com.newland.auto.pilot.collect.config;

import com.newland.auto.pilot.collect.ApplicationRunnerImpl;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.policy.AlwaysRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.net.URI;
import java.util.Map;

/**
 * @author:wk
 * @date:2020/5/29
 */
@Slf4j
public class MyWebSocketClient extends WebSocketClient {

    public MyWebSocketClient(URI serverUri, Map<String, String> headers) {
        super(serverUri, headers);
    }

    @Override
    public void onOpen(ServerHandshake arg0) {
        log.info("------ MyWebSocket onOpen ------");
    }

    @Override
    public void onClose(int arg0, String arg1, boolean arg2) {
        log.info("------ MyWebSocket onClose ------{}", arg1);
        log.info("啟用斷開重連!");

        try {
            // Spring-retry提供了RetryOperations接口的實現類RetryTemplate
            RetryTemplate template = new RetryTemplate();

            // 一直重試策略
            AlwaysRetryPolicy policy = new AlwaysRetryPolicy();

            // 設置重試策略
            template.setRetryPolicy(policy);

            // 執行
            String result = template.execute(
                    new RetryCallback<String, Exception>() {

                        public String doWithRetry(RetryContext context) throws Exception {
                            log.info("------------------------執行關閉連接重試操作--------------------------");
                            // 調用連接websocket方法
                            ApplicationRunnerImpl.connectWebsocket();

                            return "------------------------重連成功--------------------------";
                        }
                    },
                    // 當重試執行完閉，操作還未成為，那么可以通過RecoveryCallback完成一些失敗事后處理。
                    new RecoveryCallback<String>() {
                        public String recover(RetryContext context) throws Exception {

                            return "------------------------重試執行完閉，操作未完成！！--------------------------";
                        }
                    }
            );
            log.info(result);
        } catch (Exception ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    @Override
    public void onError(Exception arg0) {
        log.info("------ MyWebSocket onError ------{}", arg0);
    }

    @Override
    public void onMessage(String arg0) {
        log.info("-------- 接收到服務端數據： " + arg0 + "--------");
        try {
            WebSocketServer.sendInfo(arg0, "deal-send");
        } catch (Exception e) {
            log.info("發送消息到deal-send客戶端失敗，信息:" + e.getMessage());
        }
    }
}

 

• collect服務的websocket服務端，提供給deal-send服務連接，sendInfo方法里加入對用戶的監聽，deal-send用戶在線時發送信息，不在線時遠程調用deal-send服務的重連接口，在deal-send服務的websocket客戶端斷開連接時進行主動調用，使deal-send服務的websocket重新連接上。

WebSocketServer.java

package com.newland.auto.pilot.collect.config;


import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author:wk
 * @date:2020/6/10
 */
@Slf4j
@ServerEndpoint("/server/ws/{userId}")
@Component
public class WebSocketServer {

    /**
     * LongAdder記錄當前在線連接數,線程安全。
     */
    LongAdder onlineCounter = new LongAdder();
    /**
     * concurrent包的線程安全Set，用來存放每個客戶端對應的MyWebSocket對象。
     */
    private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**
     * 與某個客戶端的連接會話，需要通過它來給客戶端發送數據
     */
    private Session session;
    /**
     * 接收userId
     */
    private String userId = "";
    /**
     * 調用deal-send重連接口的uri
     */
    static String reConnectWebsocketUri = "http://localhost:9042/newland/reConnectWebsocket";


    /**
     * 連接建立成功調用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        log.info("onopen,this.session:" + session);
        this.userId = userId;
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            webSocketMap.put(userId, this);
            
        } else {
            webSocketMap.put(userId, this);
            
            onlineCounter.increment();
            //在線數加1
        }

        log.info("用戶連接:" + userId + ",當前在線人數為:" + onlineCounter.sum());

        try {
            sendMessage("連接成功");
        } catch (IOException e) {
            log.error("用戶:" + userId + ",網絡異常!!!!!!");
        }
    }

    /**
     * 連接關閉調用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            //從set中刪除
            onlineCounter.decrement();
        }
        log.info("用戶退出:" + userId + ",當前在線人數為:" + onlineCounter.sum());
    }

    /**
     * 收到客戶端消息后調用的方法
     *
     * @param message 客戶端發送過來的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("用戶消息:" + userId + ",報文:" + message);
    }

    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("用戶錯誤:" + this.userId + ",原因:" + error.getMessage());
    }  
108 
    /**
     * 實現服務器主動推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }


    /**
     * 發送自定義消息
     */
    public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {
        log.info("發送消息到:" + userId);
        if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
            webSocketMap.get(userId).sendMessage(message);
        } else {
            log.error("用戶：" + userId + ",不在線！");
            log.info("---------開始遠程調用重連websocket接口----------");
            RestTemplate restTemplate = new RestTemplate(new HttpsClientRequestFactory());
            ResponseEntity<String> response = restTemplate.getForEntity(reConnectWebsocketUri, String.class);

            log.info("執行請求后，返回response body:" + response.getBody());
            // 判斷返回狀態是否為200
            if (response.getStatusCodeValue() == 200) {
                log.info("返回狀態為200!");
                // 解析響應體
                String content = response.getBody();
                log.info(content);
            } else {
                log.info("返回狀態為" + response.getStatusCodeValue() + "!");
            }
        }
    }


}

 

• deal-send服務的websocket客戶端連接

WebSocketConfig.java

package com.newland.auto.pilot.deal.send.config;

import com.alibaba.fastjson.JSON;
import com.newland.auto.pilot.deal.send.alarmDeal.AlarmConvert;
import com.newland.auto.pilot.deal.send.jms.FmSendJMSMgm;
import com.newland.privatefm.alarmSend.AlarmMsg;
import com.newland.privatefm.alarmSend.SendManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.util.List;

/**
 * @author:wk
 * @date:2020/6/12
 * @Description: 配置websocket客戶端
 */

@Slf4j
@Component
public class WebSocketConfig {
    @Autowired
    private AlarmConvert alarmConvert;

    @Bean
    public WebSocketClient webSocketClient() {
        try {
            WebSocketClient webSocketClient = new WebSocketClient(new URI("ws://localhost:9041" + "/server/ws/deal-send"), new Draft_6455()) {
                @Override
                public void onOpen(ServerHandshake handshakedata) {
                    log.info("------ MyWebSocket onOpen ------");
                    log.info("FmSendJMSMgm Thread Begin ...");
                    FmSendJMSMgm sendJMSMgm = new FmSendJMSMgm();
                    if (sendJMSMgm.init()) {
                        sendJMSMgm.start();
                        log.info("FmSendJMSMgm Thread End\n");
                    } else {
                        log.error("FmSendJMSMgm Thread init failed, Program Exit!!!\n");
                        System.exit(0);
                    }
                }

                @Override
                public void onMessage(String message) {
                    log.info("-------- 接收到服務端數據： " + message + "--------");

                    if (StringUtils.isNotBlank(message)) {
                        try {
                            //解析發送的報文
                            List<AlarmMsg> alarmMsgList = alarmConvert.alarmMsgConvert(message);
                            alarmMsgList.parallelStream().forEach(alarmMsg -> {
   
                                //Jms發送數據
                                log.info("Jms發送數據:" + JSON.toJSONString(alarmMsg));
                                SendManager.sendManager.putData(alarmMsg);
                                log.info("Jms發送數據成功！");
                            });
                        } catch (Exception e) {
                            log.error(e.getMessage());
                        }
                    }

                }

                @Override
                public void onClose(int code, String reason, boolean remote) {
                    log.info("------ MyWebSocket onClose ------{}", reason);
                }

                @Override
                public void onError(Exception ex) {
                    log.info("------ MyWebSocket onError ------{}", ex);
                }
            };
            webSocketClient.connect();
            return webSocketClient;
        } catch (Exception e) {
            log.error(e.getMessage, e);
        }
        return null;
    }

}

 

• deal-send服務的遠程重連接口

ReConnectWsController.java

package com.newland.auto.pilot.deal.send.controller;

import com.newland.auto.pilot.deal.send.config.WebSocketConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

/**
 * @author:wk
 * @date:2020/9/4 調用方法進行重連websocket
 */

@Slf4j
@Controller
@ResponseBody
@RequestMapping(value = "/newland")
public class ReConnectWsController {
    @Autowired
    WebSocketConfig webSocketConfig;

    @GetMapping("/reConnectWebsocket")
    public String reConnectWebsocket() throws Exception {
        String returnMsg = "---------調用重連websocket接口成功-----------";

        log.info("---------開始調用重連websocket接口----------");
        webSocketConfig.webSocketClient().reconnect();
        log.info(returnMsg);
        return returnMsg;
    }
}

 

• collect服務websocket客戶端重連日志

[2020-09-05 11:25:18] 126481 [WebSocketTimer] INFO -c.n.a.p.c.config.MyWebSocketClient[36] - ------ MyWebSocket onClose ------The connection was closed because the other endpoint did not respond with a pong in time. For more information check: https://github.com/TooTallNate/Java-WebSocket/wiki/Lost-connection-detection 
[2020-09-05 11:25:18] 126485 [WebSocketTimer] INFO -c.n.a.p.c.config.MyWebSocketClient[37] - 啟用斷開重連! 
[2020-09-05 11:25:18] 126501 [WebSocketTimer] INFO -c.n.a.p.c.config.MyWebSocketClient[54] - ------------------------執行關閉連接重試操作-------------------------- 
[2020-09-05 11:25:18] 126505 [WebSocketTimer] INFO -c.n.a.p.c.collect.TokenCollect[31] - 開始執行請求！ 
[2020-09-05 11:25:18] 126506 [WebSocketTimer] INFO -c.n.a.p.c.collect.TokenCollect[39] - 開始使用restTemplate.exchange！！ 
[2020-09-05 11:25:18] 126705 [WebSocketTimer] INFO -c.n.a.p.c.collect.TokenCollect[41] - 執行請求后，返回response body:{"accessSession":"x-xxxxxxxxxxx","roaRand":"0xxxxxxxxxx","expires":1800,"additionalInfo":null} 
[2020-09-05 11:25:18] 126706 [WebSocketTimer] INFO -c.n.a.p.c.collect.TokenCollect[44] - 返回狀態為200! 
[2020-09-05 11:25:18] 126706 [WebSocketTimer] INFO -c.n.a.p.c.collect.TokenCollect[47] - {"accessSession":"x-xxxxxxxxxxx","roaRand":"0xxxxxxxxx","expires":1800,"additionalInfo":null} 
[2020-09-05 11:25:18] 126706 [WebSocketTimer] INFO -c.n.a.p.c.collect.TokenCollect[50] - accessToken:x-xxxxxuxxxxxx
[2020-09-05 11:25:18] 126711 [WebSocketTimer] INFO -c.n.a.p.c.collect.SubscribeCollect[46] - 開始執行請求！ 
[2020-09-05 11:25:18] 126917 [WebSocketTimer] INFO -c.n.a.p.c.collect.SubscribeCollect[49] - 執行請求后，返回response body:{"output":{"subscription-result":"ok","identifier":"9xxxxxxxx","url":"/restconf/streams/ws/v1/identifier/34xxxxx"}} 
[2020-09-05 11:25:18] 126917 [WebSocketTimer] INFO -c.n.a.p.c.collect.SubscribeCollect[52] - 返回狀態為200! 
[2020-09-05 11:25:18] 126917 [WebSocketTimer] INFO -c.n.a.p.c.collect.SubscribeCollect[55] - {"output":{"subscription-result":"ok","identifier":"95xxxxxxx","url":"/restconf/streams/ws/v1/identifier/34xxxxxxx"}} 
[2020-09-05 11:25:18] 126918 [WebSocketTimer] INFO -c.n.a.p.c.collect.SubscribeCollect[59] - url:/restconf/streams/ws/v1/identifier/34xxxxxxx 
[2020-09-05 11:25:18] 126918 [WebSocketTimer] INFO -c.n.a.p.c.ApplicationRunnerImpl[58] - protocol:TLSv1.2 
[2020-09-05 11:25:18] 126919 [WebSocketTimer] INFO -c.n.a.p.c.config.MyWebSocketClient[69] - ------------------------重連成功-------------------------- 
[2020-09-05 11:25:19] 127069 [WebSocketConnectReadThread-41] INFO -c.n.a.p.c.config.MyWebSocketClient[30] - ------ MyWebSocket onOpen ------ 
[2020-09-05 11:25:25] 133139 [WebSocketConnectReadThread-41] INFO -c.n.a.p.c.config.MyWebSocketClient[84] - -------- 接收到服務端數據： {"csn":111111,"name":"測試name","priority":"medium","occur-time":"2020-09-05T03:23:22Z","clear-time":"2020-09-05T03:26:31Z","status":"status1","category":"Power environment","domain":"domain1","source-objects":[{"id":"","type":"type1","name":"test","extend-id":"123456","device-type":"34"}],"detail":"detail1","message-type":"1","root-event-csns":[{"csn":"111","type":"1"}]}-------- 
[2020-09-05 11:25:25] 133139 [WebSocketConnectReadThread-41] INFO -c.n.a.p.c.config.WebSocketServer[136] - 發送消息到:deal-send 

 

• collect遠程調用重連websocket接口日志

[2020-09-05 11:23:33] 21128 [WebSocketConnectReadThread-33] INFO -c.n.a.p.c.config.WebSocketServer[136] - 發送消息到:deal-send 
[2020-09-05 11:23:33] 21137 [WebSocketConnectReadThread-33] ERROR-c.n.a.p.c.config.WebSocketServer[140] - 用戶：deal-send,不在線！ 
[2020-09-05 11:23:33] 21138 [WebSocketConnectReadThread-33] INFO -c.n.a.p.c.config.WebSocketServer[141] - ---------開始遠程調用重連websocket接口---------- 
[2020-09-05 11:23:33] 21573 [http-nio-9041-exec-1] INFO -o.a.c.c.C.[Tomcat].[localhost].[/][173] - Initializing Spring DispatcherServlet 'dispatcherServlet' 
[2020-09-05 11:23:33] 21574 [http-nio-9041-exec-1] INFO -o.s.web.servlet.DispatcherServlet[524] - Initializing Servlet 'dispatcherServlet' 
[2020-09-05 11:23:33] 21588 [http-nio-9041-exec-1] INFO -o.s.web.servlet.DispatcherServlet[546] - Completed initialization in 14 ms 
[2020-09-05 11:23:33] 21649 [http-nio-9041-exec-1] INFO -c.n.a.p.c.config.WebSocketServer[49] - onopen,this.session:org.apache.tomcat.websocket.WsSession@32f5b990 
[2020-09-05 11:23:33] 21650 [http-nio-9041-exec-1] INFO -c.n.a.p.c.config.WebSocketServer[62] - 用戶連接:deal-send,當前在線人數為:1 

 

• deal-send服務被遠程調用重連websocket接口日志

[2020-09-05 11:23:08] 87821 [WebSocketConnectReadThread-19] INFO -c.n.a.p.d.s.config.WebSocketConfig[74] - ------ MyWebSocket onClose ------ 
[2020-09-05 11:23:33] 112532 [http-nio-9042-exec-1] INFO -o.a.c.c.C.[Tomcat].[localhost].[/][173] - Initializing Spring DispatcherServlet 'dispatcherServlet' 
[2020-09-05 11:23:33] 112533 [http-nio-9042-exec-1] INFO -o.s.web.servlet.DispatcherServlet[524] - Initializing Servlet 'dispatcherServlet' 
[2020-09-05 11:23:33] 112556 [http-nio-9042-exec-1] INFO -o.s.web.servlet.DispatcherServlet[546] - Completed initialization in 23 ms 
[2020-09-05 11:23:33] 112630 [http-nio-9042-exec-1] INFO -c.n.a.p.d.s.c.ReConnectWsController[28] - ---------開始調用重連websocket接口---------- 
[2020-09-05 11:23:33] 112828 [WebSocketConnectReadThread-43] INFO -c.n.a.p.d.s.config.WebSocketConfig[38] - ------ MyWebSocket onOpen ------