分布式下的WebSocket解決方案是什么

本篇內容主要講解“分布式下的WebSocket解決方案是什么”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“分布式下的WebSocket解決方案是什么”吧!

WebSocket單體應用介紹

在介紹分布式集群之前，我們先來看一下王子的WebSocket代碼實現，先來看java后端代碼如下：

import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject;import java.io.IOException; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @ServerEndpoint("/webSocket/{key}") public class WebSocket {     private static int onlineCount = 0;     /**      * 存儲連接的客戶端      */     private static Map<String, WebSocket> clients = new ConcurrentHashMap<String, WebSocket>();     private Session session;     /**      * 發送的目標科室code      */     private String key;     @OnOpen     public void onOpen(@PathParam("key") String key, Session session) throws IOException {         this.key = key;         this.session = session;         if (!clients.containsKey(key)) {             addOnlineCount();        }        clients.put(key, this);         Log.info(key+"已連接消息服務！");     }    @OnClose     public void onClose() throws IOException {         clients.remove(key);        subOnlineCount();    }    @OnMessage     public void onMessage(String message) throws IOException {         if(message.equals("ping")){             return ;         }        JSONObject jsonTo = JSON.parseObject(message);        String mes = (String) jsonTo.get("message");         if (!jsonTo.get("to").equals("All")){             sendMessageTo(mes, jsonTo.get("to").toString());         }else{             sendMessageAll(mes);        }    }    @OnError     public void onError(Session session, Throwable error) {         error.printStackTrace();    }    private void sendMessageTo(String message, String To) throws IOException {         for (WebSocket item : clients.values()) {             if (item.key.contains(To) )                 item.session.getAsyncRemote().sendText(message);        }    }    private void sendMessageAll(String message) throws IOException {         for (WebSocket item : clients.values()) {             item.session.getAsyncRemote().sendText(message);        }    }    public static synchronized int getOnlineCount() {         return onlineCount;     }    public static synchronized void addOnlineCount() {         WebSocket.onlineCount++;    }    public static synchronized void subOnlineCount() {         WebSocket.onlineCount--;    }    public static synchronized Map<String, WebSocket> getClients() {         return clients;     }}

示例代碼中并沒有使用Spring，用的是原生的java web編寫的，簡單和大家介紹一下里面的方法。

• onOpen：在客戶端與WebSocket服務連接時觸發方法執行

• onClose：在客戶端與WebSocket連接斷開的時候觸發執行

• onMessage：在接收到客戶端發送的消息時觸發執行

• onError：在發生錯誤時觸發執行

可以看到，在onMessage方法中，我們直接根據客戶端發送的消息，進行消息的轉發功能，這樣在單體消息服務中是沒有問題的。

再來看一下js代碼

var host = document.location.host;     // 獲得當前登錄科室    var deptCodes='${sessionScope.$UserContext.departmentID}';     deptCodes=deptCodes.replace(/[\[|\]|\s]+/g, "");     var key = '${sessionScope.$UserContext.userID}'+deptCodes;     var lockReconnect = false;  //避免ws重復連接     var ws = null;          // 判斷當前瀏覽器是否支持WebSocket    var wsUrl = 'ws://' + host + '/webSocket/'+ key;     createWebSocket(wsUrl);   //連接ws    function createWebSocket(url) {         try{            if('WebSocket' in window){                 ws = new WebSocket(url);            }else if('MozWebSocket' in window){                   ws = new MozWebSocket(url);            }else{                   layer.alert("您的瀏覽器不支持websocket協議,建議使用新版谷歌、火狐等瀏覽器，請勿使用IE10以下瀏覽器，360瀏覽器請使用極速模式，不要使用兼容模式！");              }            initEventHandle();        }catch(e){            reconnect(url);            console.log(e);         }         }    function initEventHandle() {         ws.onclose = function () {             reconnect(wsUrl);            console.log("llws連接關閉!"+new Date().toUTCString());         };        ws.onerror = function () {             reconnect(wsUrl);            console.log("llws連接錯誤!");         };        ws.onopen = function () {             heartCheck.reset().start();      //心跳檢測重置            console.log("llws連接成功!"+new Date().toUTCString());         };        ws.onmessage = function (event) {    //如果獲取到消息，心跳檢測重置             heartCheck.reset().start();      //拿到任何消息都說明當前連接是正常的//接收到消息實際業務處理　 ...        };    }    // 監聽窗口關閉事件，當窗口關閉時，主動去關閉websocket連接，防止連接還沒斷開就關閉窗口，server端會拋異常。    window.onbeforeunload = function() {         ws.close();     }      function reconnect(url) {         if(lockReconnect) return;         lockReconnect = true;         setTimeout(function () {     //沒連接上會一直重連，設置延遲避免請求過多             createWebSocket(url);            lockReconnect = false;         }, 2000);     }    //心跳檢測    var heartCheck = {        timeout: 300000,        //5分鐘發一次心跳         timeoutObj: null,        serverTimeoutObj: null,        reset: function(){             clearTimeout(this.timeoutObj);            clearTimeout(this.serverTimeoutObj);            return this;         },        start: function(){             var self = this;            this.timeoutObj = setTimeout(function(){                 //這里發送一個心跳，后端收到后，返回一個心跳消息，                //onmessage拿到返回的心跳就說明連接正常                ws.send("ping");                 console.log("ping!")                 self.serverTimeoutObj = setTimeout(function(){//如果超過一定時間還沒重置，說明后端主動斷開了                     ws.close();     //如果onclose會執行reconnect，我們執行ws.close()就行了.如果直接執行reconnect 會觸發onclose導致重連兩次                 }, self.timeout)            }, this.timeout)        }}

js部分使用的是原生H5編寫的，如果為了更好的兼容瀏覽器，也可以使用SockJS，有興趣小伙伴們可以自行百度。

接下來我們就手動的優化代碼，實現WebSocket對分布式架構的支持。

解決方案的思考

現在我們已經了解單體應用下的代碼結構，也清楚了WebSocket在分布式環境下面臨的問題，那么是時候思考一下如何能夠解決這個問題了。

我們先來看一看發生這個問題的根本原因是什么。

簡單思考一下就能明白，單體應用下只有一臺服務器，所有的客戶端連接的都是這一臺消息服務器，所以當發布消息者發送消息時，所有的客戶端其實已經全部與這臺服務器建立了連接，直接群發消息就可以了。

換成分布式系統后，假如我們有兩臺消息服務器，那么客戶端通過Nginx負載均衡后，就會有一部分連接到其中一臺服務器，另一部分連接到另一臺服務器，所以發布消息者發送消息時，只會發送到其中的一臺服務器上，而這臺消息服務器就可以執行群發操作，但問題是，另一臺服務器并不知道這件事，也就無法發送消息了。

現在我們知道了根本原因是生產消息時，只有一臺消息服務器能夠感知到，所以我們只要讓另一臺消息服務器也能感知到就可以了，這樣感知到之后，它就可以群發消息給連接到它上邊的客戶端了。

那么什么方法可以實現這種功能呢，王子很快想到了引入消息中間件，并使用它的發布訂閱模式來通知所有消息服務器就可以了。

引入RabbitMQ解決分布式下的WebSocket問題

在消息中間件的選擇上，王子選擇了RabbitMQ，原因是它的搭建比較簡單，功能也很強大，而且我們只是用到它群發消息的功能。

RabbitMQ有一個廣播模式(fanout)，我們使用的就是這種模式。

首先我們寫一個RabbitMQ的連接類：

import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class RabbitMQUtil {     private static Connection connection;     /**      * 與rabbitmq建立連接      * @return      */     public static Connection getConnection() {         if (connection != null&&connection.isOpen()) {             return connection;         }        ConnectionFactory factory = new ConnectionFactory();         factory.setVirtualHost("/");         factory.setHost("192.168.220.110"); // 用的是虛擬IP地址         factory.setPort(5672);         factory.setUsername("guest");         factory.setPassword("guest");         try {             connection = factory.newConnection();         } catch (IOException e) {             e.printStackTrace();         } catch (TimeoutException e) {             e.printStackTrace();         }         return connection;     } }

這個類沒什么說的，就是獲取MQ連接的一個工廠類。

然后按照我們的思路，就是每次服務器啟動的時候，都會創建一個MQ的消費者監聽MQ的消息，王子這里測試使用的是Servlet的監聽器，如下：

import javax.servlet.ServletContextEvent; import javax.servlet.ServletContextListener; public class InitListener implements ServletContextListener {     @Override     public void contextInitialized(ServletContextEvent servletContextEvent) {         WebSocket.init();    }    @Override     public void contextDestroyed(ServletContextEvent servletContextEvent) {     }}

記得要在Web.xml中配置監聽器信息

<?xml version="1.0" encoding="UTF-8"?> <web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"          xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_4_0.xsd"          version="4.0">     <listener>         <listener-class>InitListener</listener-class>     </listener> </web-app>

WebSocket中增加init方法，作為MQ消費者部分

public  static void init() {         try {            Connection connection = RabbitMQUtil.getConnection();            Channel channel = connection.createChannel();            //交換機聲明（參數為：交換機名稱；交換機類型）             channel.exchangeDeclare("fanoutLogs",BuiltinExchangeType.FANOUT);             //獲取一個臨時隊列             String queueName = channel.queueDeclare().getQueue();            //隊列與交換機綁定（參數為：隊列名稱；交換機名稱；routingKey忽略）             channel.queueBind(queueName,"fanoutLogs","");             //這里重寫了DefaultConsumer的handleDelivery方法，因為發送的時候對消息進行了getByte()，在這里要重新組裝成String             Consumer consumer = new DefaultConsumer(channel) {                @Override                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                    super.handleDelivery(consumerTag, envelope, properties, body);                     String message = new String(body,"UTF-8");                     System.out.println(message);//這里可以使用WebSocket通過消息內容發送消息給對應的客戶端                 }            };            //聲明隊列中被消費掉的消息（參數為：隊列名稱；消息是否自動確認;consumer主體）             channel.basicConsume(queueName,true,consumer);             //這里不能關閉連接，調用了消費方法后，消費者會一直連接著rabbitMQ等待消費         } catch (IOException e) {            e.printStackTrace();        }    }

同時在接收到消息時，不是直接通過WebSocket發送消息給對應客戶端，而是發送消息給MQ，這樣如果消息服務器有多個，就都會從MQ中獲得消息，之后通過獲取的消息內容再使用WebSocket推送給對應的客戶端就可以了。

WebSocket的onMessage方法增加內容如下：

try {             //嘗試獲取一個連接             Connection connection = RabbitMQUtil.getConnection();            //嘗試創建一個channel             Channel channel = connection.createChannel();            //聲明交換機（參數為：交換機名稱; 交換機類型，廣播模式）             channel.exchangeDeclare("fanoutLogs", BuiltinExchangeType.FANOUT);             //消息發布（參數為：交換機名稱; routingKey，忽略。在廣播模式中，生產者聲明交換機的名稱和類型即可）             channel.basicPublish("fanoutLogs","", null,msg.getBytes("UTF-8"));             System.out.println("發布消息");             channel.close();        } catch (IOException |TimeoutException e) {             e.printStackTrace();         }

增加后刪除掉原來的Websocket推送部分代碼。

到此，相信大家對“分布式下的WebSocket解決方案是什么”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！