前言
最近本人搭建了个我的项目，我的项目自身很简略，然而外面有应用WebSocket进行音讯揭示的性能，大体状况是这样的。
公布音讯者在零碎中发送音讯，实时的把音讯推送给对应的一个部门下的所有人。
这外面如果是单机利用的状况时，咱们能够通过部门的id和用户的id组成一个惟一的key，与应用服务器建设WebSocket长连贯，而后就能够接管到公布音讯者发送的音讯了。
然而真正把我的项目利用于生产环境中时，咱们是不可能就部署一个单机利用的，而是要部署一个集群。
所以我通过Nginx+两台Tomcat搭建了一个简略的负载平衡集群，作为测试应用
然而问题呈现了，咱们的客户端浏览器只会与一台服务器建设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;@OnOpenpublic 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+"已连贯音讯服务！");}@OnClosepublic void onClose() throws IOException { clients.remove(key); subOnlineCount();}@OnMessagepublic 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); }}@OnErrorpublic 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; // 判断以后浏览器是否反对WebSocketvar wsUrl = 'ws://' + host + '/webSocket/'+ key;createWebSocket(wsUrl); //连贯wsfunction 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 {
@Overridepublic void contextInitialized(ServletContextEvent servletContextEvent) { WebSocket.init();}@Overridepublic 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的推送音讯问题。
咱们次要是引入了RabbitMQ，通过RabbitMQ的公布订阅模式，让每个音讯服务器启动的时候都去订阅音讯，而无论哪台音讯服务器在发送音讯的时候都会发送给MQ，这样每台音讯服务器就都会感知到发送音讯的事件，从而再通过Websocket发送给客户端。
大体流程就是这样，那么小伙伴们有没有想过，如果RabbitMQ挂掉了几分钟，之后重启了，消费者是否能够从新连贯到RabbitMQ？是否还能失常接管音讯呢？
生产环境下，这个问题是必须思考的。
这里曾经测试过，消费者是反对主动重连的，所以咱们能够释怀的应用这套架构来解决此问题。
本文到这里就完结了，欢送各位小伙伴点赞文章留言探讨，一起学习，一起提高。