• WebSocket 大家应该是再熟悉不过了，如果是单体应用确实不会有什么问题，但是当我们的项目使用微服务架构时，就可能会存在问题

• 比如 服务A 有两个实例 A1 和 A2 ，前端的 WebSocket 客户端 C 通过网关的负载均衡连到了 A1 ，这个时候当 A2 触发消息发送的逻辑，需要将某个消息发送给所有的客户端时， C 就接受不到消息

• 这个时候我们很快就能想到一种最简单的解决方案，就是把 A2 的消息转发给 A1 ， A1 再把消息发送给 C ，这样 C 就能收到 A2 发送的消息了

基于这个思路，我实现了一个库，一个配置注解搞定一切

接下来让我们看看这个库的用法

首先我们需要在启动类上添加一个注解 @EnableWebSocketLoadBalanceConcept

@EnableWebSocketLoadBalanceConcept
@EnableDiscoveryClient
@SpringBootApplication
public class AServiceApplication {
 public static void main(String[] args) {
 SpringApplication.run(AServiceApplication.class, args);
接着我们在需要发送消息的地方注入WebSocketLoadBalanceConcept就可以愉快的跨实例发消息啦
 
@RestController
@RequestMapping("/ws")
public class WsController {
 @Autowired
 private WebSocketLoadBalanceConcept concept;
 @RequestMapping("/send")
 public void send(@RequestParam String msg) {
 concept.send(msg);
 
是不是很简单，有没有觉得比自己集成单体应用的WebSocket还要简单！当你的同事还在头疼要实现手动转发时你已经通过一个配置注解实现了功能并开始泡茶喝。你的同事肯定对你刮目相看啊（又能开始摸鱼了）不知道大家看了之后是不是对具体实现已经有了一些思路呢
 
 
接下来我就来讲讲这个库的实现流程
 
 
 
其实我之前有专门针对WebSocket实现过类似功能的模块，只是当时的一些场景都是基于项目定死的，所以相对来说实现比较简单，但是过于定制化不好扩展
 
 
有一天在和我的一个前同事聊天的过程中得知，他们在考虑让设备和服务直连，并且服务要部署成多实例
 
 
设备和服务直连无非就是通过TCP这种长连接来实现，可以使用缓存来保存连接和服务地址的映射关系来实现点对点转发的功能需求
 
 
听到这里，是不是感觉似曾相识？当时就有一道光穿过我的脑瓜子，真相只有一个！这不就和WebSocket在集群模式下的问题一样么
 
 
于是我从原来针对WebSocket的思考，变成了对各种长连接的思考，最终我将这个问题抽象成了：长连接的集群方案
 而不管是WebSocket还是TCP都是长连接的一种具体实现，所以我们可以抽象一个顶级接口Connection，然后实现WebSocketConnection或者是TCPConnection。其实从抽象的角度来说不仅仅是长连接，短连接也在我们的抽象范围之内，只不过类似HTTP等协议并不存在上述的问题，但是并不妨碍你实现一个HTTPConnection用于转发消息，所以大家不要被先入为主的思维束缚住了
 
 
 
之前讲到，这个库的主要思路就是将消息转发给其他的服务实例来达到一个单播或广播的效果，所以消息转发的设计就非常重要了。首先消息转发需要凭借一些支持数据交互的技术手段，比如HTTP，MQ，TCP，WebSocket。说到这里…大家是不是…你TM原来自己就能搞定啊（掀桌）长连接不就是用来交互数据的吗，所以完全可以自给自足啊,于是就有一个精妙的想法在我脑子里形成：
 
  
如果每个服务实例都把自己作为一个客户端，连接到其他服务上呢？
WebSocket的场景下，我们将当前服务实例作为一个WebSocket客户端去连接其他服务实例的WebSocket服务端
TCP的场景下，我们将当前服务实例作为一个TCP的客户端去连接其他服务实例的TCP服务端
这样其他服务实例就可以把消息发到这些伪装的客户端上，当服务实例上伪装的客户端接收到消息之后就可以再转发给自己管理的真正的客户端
 
 
所以我们首先需要先让服务实例之间相互连接上
 
 
让我们来看看互相建立连接是怎么设计的
 
我定义了一个ConnectionSubscriber的接口，大家可以理解为我们的服务实例要去订阅监听其他服务发送的消息
 
同时提供了默认实现，就是基于自身的协议进行连接和消息的发送
 
当然也能够灵活的支持其他方式，只需要自定义一个ConnectionSubscriber就可以了，如果使用MQ的方式就可以实现一个MQConnectionSubscriber或者使用HTTP就可以实现一个HTTPConnectionSubscriber
 
只不过使用自身的协议就可以不用依赖其他的库或是中间件了，当然如果你对消息的丢失率有比较严格的要求也可以使用MQ作为消息转发的中介，而以我之前参与过的项目来说，一般普通的WebSocket场景基本上还是能忍受一定的丢失率的
 
获取服务实例信息
 
那么我们怎么知道要去连接哪些实例呢
 
我定义了一个ConnectionServerManager的接口用来管理服务信息
 
当然我们完全可以自己实现一个，比如通过配置文件来配置服务实例信息
 
不过我们有更方便的方式，那就是依赖Spring Cloud的服务发现组件了，不管是Eureka还是Nacos还是其他的注册中心相当于都支持了，这就是抽象的魅力啊
 
我们可以通过DiscoveryClient#getInstances(Registration.getServiceId())来获得所有的实例，排除掉自身就是需要连接的服务实例了
 
当我们的服务实例连接上其他的服务实例之后，发送一个自身实例信息的消息过去，其他的服务实例接收到对应的消息之后反过来连接我们的服务实例，保证一定的连接及时性，这样双方的连接就搭建起来了，可以互相转发消息了
 
同时我还添加了心跳检测和自动重连，当一段时间没有收到心跳回复后就会断开连接，并且每隔一段时间就会重新查询一遍实例信息，如果发现存在某个服务实例没有对应的连接，就会重新进行连接，这样就能在某些偶尔网络不好的情况下有一定的容错
 
到目前为止，我们基本的框架已经建立了，当我们启动服务之后，服务间就会自动建立连接
 
连接区分和管理
 
基于上述的思路，我们肯定需要区分真实的客户端和用来转发的客户端
 
于是我就把这些连接做了一个分类
 
类别
说明
Client
普通的连接
Subscriber
服务实例伪装的连接，用于接受需要转发的消息
Observable
服务实例伪装的连接，用于发送需要转发的消息
 
然后对于这些连接进行一个统一的管理
 
通过连接工厂ConnectionFactory我们可以将任意的连接适配成Connection对象，并实现各种连接间的消息转发
 
每个连接都会配置一个MessageEncoder和MessageDecoder用于消息的编码和解码，而且不同类别的连接对应的编码器和解码器肯定是不一样的，比如转发的消息和发给真实客户端的消息很大程度上都是有区别的，所以额外定义了一个MessageCodecAdapter用来适配不同类型的编解码器，也能让大家在自定义时方便管理
 
现在当我们发送某条消息之后，消息就会被转发到其他的服务实例，所有的客户端就都能收到了
 
不对啊，在有些情况下我们不想让所有客户端都收到啊，能不能我们想让谁收到就让谁收到啊
 
真麻烦，来，我把所有的连接都给你，你自己选吧
 
我们需要在消息发送时确定发送给哪些连接
 
于是我就定义了一个连接选择器ConnectionSelector
 
每次要发送消息的时候，我都会匹配一个连接选择器，然后通过选择器来获得需要发送消息的连接，而我们可以通过自定义连接选择器来实现我们消息的精准发送
 
这里其实就是我为什么会取名WebSocketLoadBalanceConcept的原因，为什么要叫LoadBalance呢
 
Ribbon通过IRule来选择一个Server
 
我通过ConnectionSelector来选择一个Connection集合
 
是不是有异曲同工之妙
 
继续来说自定义选择器
 
准备工作：
 
 
我们的Connection有一个metadata字段用于存放自定义属性
 
 
我们的Message有一个headers字段用于存放消息头
 
 
给指定用户发送消息
 
 
很多场景下我们需要给指定的用户发送消息
 
 
首先当客户端连接上来时，可以通过参数或者主动发送一个消息将userId发给服务端，然后服务端将得到的userId存在Connection的metadata中
 
 
接着我们给需要发送的Message添加一个header，将对应的userId作为消息头
 
 
这样我们就可以自定义一个连接选择器通过判断Message是否包含userId消息头来作为匹配的条件，当Message的headers中存在userId时，对Connection中的metadata进行userId的匹配来筛选需要发送消息的连接
 
 
由于userId是唯一的，当我们自身服务连上来的客户端中已经匹配到就不需要再转发了，如果没有匹配到就通过其他服务实例的客户端进行消息转发
 
 
库中已经实现了对应的UserSelector和UserMessage，可以使用配置开启并通过在连接路径上添加userId参数来标记用户
 
 
当然我们也可以借用缓存来精确的判断需不需要转发或者是需要转发给哪几个服务，把userId和服务的instanceId等一些具有唯一性的数据缓存在Redis中，当给用户发送消息时，从Redis中获得用户对应的服务实例的instanceId或是具有唯一性的数据，如果经过匹配就是当前服务就可以直接下发，如果是其他服务就转发给那个对应的服务就行了
 
 
给指定路径发送消息
 
 
还有一种场景也比较常见就是类似主题订阅，如订阅设备状态更新的数据，就要给每一个对应路径的连接发送消息了
 
 
我们可以使用不同的路径来表示不同主题，然后自定义一个连接选择器来匹配连接的路径和消息头中指定的路径
 
 
当然库中也已经实现了对应的PathSelector和PathMessage，可以通过配置开启
 
 
 
最后请允许我发表一点对于抽象的拙见
 
 
抽象其实就和 “道生一，一生二，二生三，三生万物” 一样，根据你的顶级接口（也就是核心功能）不断的向外展开，你的顶级接口就是道（狭义的来讲）
 
 
以这个库为例，ConnectionLoadBalanceConcept就是这个库的道，他的核心功能就是发送消息，至于怎么发，发给谁，不确定，像是一个混沌的状态
 
 
那么什么是一，二，三呢，我们发送消息需要载体于是就有了Connection和Message，我们需要对Connection进行管理于是就有了ConnectionRepository， 我们需要转发消息于是就有了ConnectionSubscriber等等
 
 
而万物就像是具体的实现，是能落实的，基于Spring Cloud服务发现的连接管理器DiscoveryConnectionServerManager，基于路径的连接选择器PathSelector，基于Reactive的WebSocket连接ReactiveWebSocketConnection
 
 
https://github.com/Linyuzai/concept/wiki/Concept-WebSocket-LoadBalance
 
原文地址：https://juejin.cn/post/7105931939407740965#comment
                                    最近做项目时遇到了需要多用户之间通信的问题，涉及到了WebSocket握手请求，以及集群中WebSocket Session共享的问题。期间我经过了几天的研究，总结出了几个实现分布式WebSocket集群的办法，从zuul到spring cloud gateway的不同尝试，总结出了这篇文章，希望能帮助到某些人，并且能一起分享这方面的想法与研究。以下是我的场景描述：4台服务器。其中只有一台服务器具备ssl认证域名，一台redis+mysql服务器，两台应用服务器（集群）
                                    探秘Spring Cloud WebSocket集群：高性能实时通信解决方案
项目地址:https://gitcode.com/Lonor/websocket-cluster
WebSocket，作为一种在Web上实现全双工通信的协议，已经广泛应用于实时数据传输，如聊天应用、在线游戏、股票交易等场景。如今，随着微服务架构的发展，如何在分布式环境下优雅地实现WebSocket成了开发者关注的重要议题...
                                    点击关注公众号，Java干货及时送达????前言WebSocket是一种在网络应用程序中，使客户度端和服务器之间可以进行双向通信的协议。它允许数据可以在建立连接后进行实时交换，而不必依赖传统的HTTP请求-响应模式。WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性...
                                    Websocket是一种保持长连接的技术，并且是双向的，适合于频繁和较大流量的双向通讯场景，是服务器推送消息功能的最佳实践。而实现websocket的最佳方式，就是netty，是一款基于 NIO（Nonblocking IO，非阻塞IO）开发异步事件驱动的高性能网络通信框架，nio使用了select模型（多路复用器技术），从而使得系统在单线程的情况下可以同时处理多个客户端请求。Netty使用了Reactor模型，Reactor模型有三种多线程模型，netty是在主从 Reactor 多线程模型上的改进
                                    目录一、服务端配置1、引入依赖包2、编写Config类进行websocket的配置3、编写拦截器WebSocketInterceptor4、编写消息的接收和发送的Controller二、客户端配置1、引入SockJS以及stompjs2、编写公用的websocket.js供其他地方调用3、业务代码中引用websocket进行消息发送和接收三、Spring cloud gateway网关配置1、修改gateway网关的配置文件2、在网关中添加全局过滤器四、前端代理的配置五、问题记录
一、服务端配置
整个服务端
                                    问题起因最近做项目时遇到了需要多用户之间通信的问题，涉及到了WebSocket握手请求，以及集群中WebSocket Session共享的问题。期间我经过了几天的研究，总结出了几个实现分布式WebSocket集群的办法，从zuul到spring cloud gateway的不同尝试，总结出了这篇文章，希望能帮助到某些人，并且能一起分享这方面的想法与研究。以下是我的场景描述资源：4台服务器。其中只有...
(1)、WebSocketServer类
import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.List;
import javax.websocket.OnClose;
import javax.websocket.OnMessage;
impo...
                                    话不多说，直接上代码：1.首先创建websocket的配置类2.创建websocket核心服务这样其实就已经可以连接了，但在实际运用中，会出现一种情况，如图：状态码明明显示的200，但却连接之后却立即断掉了，此时很有可能是被某个过滤器拦截掉，我查看了自己项目中的过滤器，发现都放行了，却在下一步直接断开；在过滤器与websocket服务之间只有一个，那就是用户认证。所以我先将websocket的认证给关掉。如图：我用的是spring-security的配置连接成功。
                                    最近项目中有一个站内信推送的功能需要优化，之前一直都是前端不断定时去查询新消息，造成很多不必要的资源浪费，现在想使用websocket保持长连接，有新消息的时候通过websocket发送特定的标识给前端，前端收到后再去调用查询新消息的接口。Netty 是一个开源的、异步的、事件驱动的网络应用程序框架，用于快速开发高性能的网络和分布式系统。：Netty的设计采用了组件化和可重用的原则，它提供了一系列的易于扩展和替换的组件，如编解码器、处理器、Handler等，开发人员可以根据需要组合和定制这些组件。
                                    作为一名程序员，工作也7、8年了，这是我第一次写文章，实在是惭愧。之前因工作需要一直都是百度、google，而没有想过静下心来自己研究原理、源码，导致现在还是一只菜鸟。。。
好了。言归正传，本次实践的话是基于nacos作为配置、注册中心的，各位大佬可以去了解一下nacos的官网，虽然最近这个组件爆出了漏洞，但不可否认其功能还真的是很强大的，附上官网地址：https://nacos.io/zh-cn/
项目使用jdk1.8进行开发。使用maven做jar包管理，分了很多模块子项目，包括基础服务的commo
                                    WebSocket大家应该是再熟悉不过了，如果是单体应用确实不会有什么问题，但是当我们的项目使用微服务架构时，就可能会存在问题比如服务A有两个实例A1和A2，前端的WebSocket客户端C通过网关的负载均衡连到了A1，这个时候当A2触发消息发送的逻辑，需要将某个消息发送给所有的客户端时，C就接受不到消息这个时候我们很快就能想到一种最简单的解决方案，就是把A2的消息转发给A1，A1再把消息发送给C，这样C就能收到A2发送的消息了基于这个思路，我实现了一个库，一个配置注解搞定一切❝。