一个注解实现 WebSocket 集群方案，这样玩才爽！ 上

• 介绍
• 用法

• 抽象思路
• 转发思路
• 连接流程
• 消息发送

• 结束

介绍

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

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

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

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

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

• 项目地址：https://github.com/YunaiV/ruoyi-vue-pro
• 视频教程：https://doc.iocoder.cn/video/

用法

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

首先我们需要在启动类上添加一个注解@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())来获得所有的实例，排除掉自身就是需要连接的服务实例了

当我们的服务实例连接上其他的服务实例之后，发送一个自身实例信息的消息过去，其他的服务实例接收到对应的消息之后反过来连接我们的服务实例，保证一定的连接及时性，这样双方的连接就搭建起来了，可以互相转发消息了

同时我还添加了心跳检测和自动重连，当一段时间没有收到心跳回复后就会断开连接，并且每隔一段时间就会重新查询一遍实例信息，如果发现存在某个服务实例没有对应的连接，就会重新进行连接，这样就能在某些偶尔网络不好的情况下有一定的容错

到目前为止，我们基本的框架已经建立了，当我们启动服务之后，服务间就会自动建立连接

连接区分和管理

基于上述的思路，我们肯定需要区分真实的客户端和用来转发的客户端

于是我就把这些连接做了一个分类