⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 juejin.cn/post/7306451559928348709 「kpioneer」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前言

WebSocket是一种在网络应用程序中，使客户度端和服务器之间可以进行双向通信的协议。它允许数据可以在建立连接后进行实时交换，而不必依赖传统的HTTP请求-响应模式。WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。

在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

方案一：广播方案

Step1: 客户端连接到某个Websocket Server，在该websocket Server中建立userid和session的绑定关系
Step2: 其它服务或者客户端通过MQ广播消息所有Websocket Server（消息体中带有userid）
Step3: 所有Websocket Server 根据客户端userid找到对应session，只有存在userid和session的绑定关系的Websocket Server才发送消息到客户端

代码演示

1.Websocket Server 建立userid和session的绑定关系

@ServerEndpoint("/websocket/{businessType}/{userId}")
@Component
public class WebSocketServer {
    /**
     * 若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识
     * 注意：allSession 只记录当前机器的 客户端连接，不是所有session连接
     */

    public static ConcurrentHashMap<String, Session> allSession = new ConcurrentHashMap<>();

    @Resource
    private RedisService redisService;

    /**
     * 连接建立成功调用的方法
     *
     * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    @OnOpen
    public void onOpen(@PathParam(value = "businessType") String businessType, @PathParam(value = "userId") String userId, Session session, EndpointConfig config) {
        if (StringUtils.isEmpty(userId)) {
            return;
        }
        /**
         * 加入到本地map
         */
        allSession.put(userId, session);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(@PathParam(value = "userId") String userId, Session session) {
        if (StringUtils.isNotEmpty(userId)) {
            allSession.remove(userId);
        }
    }


    /**
     * 发生错误时调用
     *
     * @param
     * @param
     */
    @OnError
    public void onError(@PathParam(value = "userId") String userId, Session session, Throwable error) {
    }


    /**
     * 用户id
     *
     * @param userId
     * @param message
     */
    public void sendMessageToOneUser(Integer userId, String message, String msgId) {
        if (userId == null) {
            return;
        }
        Session session = allSession.get(String.valueOf(userId));
        if (session != null) {
         //所有Websocket Server 根据客户端userid找到对应session， 只有存在userid和session的绑定关系的Websocket Server才发送消息到客户端
          session.getAsyncRemote().sendText(message);
        } else {
            System.err.println("session为空");
            allSession.remove(userId + "");
        }
    }
}

2.所有Websocket Server 接收消息并处理

@Component
@RequiredArgsConstructor
public class CreateOrderConsumer implements BaseConsumer {


    private final WebSocketServer webSocketServer;


    @Override
    public Action handleMessage(Message message) {
        CreateOrderMessage createOrderMessage = JSON.parseObject(message.getBody(), LinkCreateOrderMessage.class);

        try {
           //业务校验省略...
           //调用WebSocketServer的sendMessageToOneUser方法，里面根据客户端userid找到对应session， 只有存在userid和session的绑定关系的Websocket Server才发送消息到客户端
            webSocketServer.sendMessageToOneUser(createOrderMessage.getUserId(), JSON.toJSONString(linkActionRes),message.getMsgID());
        } catch (Exception e) {
            e.printStackTrace();
            return Action.ReconsumeLater;
        }
        return Action.CommitMessage;
    }
}

方案二：目标询址方案（推荐）

Id标识有两种实现形式：

为唯一的服务名：每一个WebSocketServer生成唯一的服务名（serviceName="XXX-" + IdUtil.oneId()）并注册到naocs服务组册中心，uesrid与其绑定，服务适用方使用Feign 或其它RPC调用http://{serviceName}/xxx/xxx到指定WebSocketServer
为唯一的IP+端口:每一个WebSocketServer 获取自己IP+端口，uesrid与其绑定，服务调用方使用该IP+端口

代码演示(唯一Id为唯一的服务名的形式)

1.绑定userid和服务名唯一Id的关系(以ApplicationName形式为例)

@SpringBootApplication
public class WsApplication  {

    public static void main(String[] args) {
        //动态服务名
        System.setProperty("myApplicationName", "WS-" + IdUtil.oneId());
        SpringApplication.run(WsApplication.class, args);
    }
}
spring:
  application:
    #随机名字，做ws集群使用
    name: ${myApplicationName}
@ServerEndpoint("/websocket/{businessType}/{userId}")
@Component
public class WebSocketServer {
    /**
     * 若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识
     * 注意：allSession 只记录当前机器的 客户端连接，不是所有session连接
     */

    public static ConcurrentHashMap<String, Session> allSession = new ConcurrentHashMap<>();
    /**
     *
     */
    private String myApplicationName = System.getProperty("myApplicationName");
    @Resource
    private RedisService redisService;

    /**
     * 连接建立成功调用的方法
     * 关键代码
     * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    @OnOpen
    public void onOpen(@PathParam(value = "businessType") String businessType, @PathParam(value = "userId") String userId, Session session, EndpointConfig config) {
        if (StringUtils.isEmpty(userId)) {
            return;
        }
        /**
         * 加入到本地map
         */
        allSession.put(userId, session);
        //绑定userid和服务名唯一Id的关系
        redisService.hset(WS_MAPPING, userId + "", myApplicationName);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(@PathParam(value = "userId") String userId, Session session) {
        if (StringUtils.isNotEmpty(userId)) {
            allSession.remove(userId);
        }
    }

    /**
     * 发生错误时调用
     *
     * @param
     * @param
     */
    @OnError
    public void onError(@PathParam(value = "userId") String userId, Session session, Throwable error) {
    }

    /**
     * 用户id
     *
     * @param userId
     * @param message
     */
    public void sendMessageToOneUser(Integer userId, String message) {
        if (userId == null) {
            return;
        }
        Session session = allSession.get(String.valueOf(userId));
        if (session != null) {
         //所有Websocket Server 根据客户端userid找到对应session， 只有存在userid和session的绑定关系的Websocket Server才发送消息到客户端
          session.getAsyncRemote().sendText(message);
        } else {
            System.err.println("session为空");
            allSession.remove(userId + "");
        }
    }
}

2.Websocket Server提供的调用接口

@RestController
@RequestMapping("push")
public class  WebSocketPushController {


    @PostMapping("{userId}")
    public void pushMessage(@PathVariable Long userId, @RequestBody Object message) {
            webSocketServer.sendMessageToOneUser(userId, message);
    }

}

3.调用方通过nacos调用目标Websocket Server

//业省略
MyApplicationName myApplicationName =  redisService.hget(WS_MAPPING, userId + "");

Feign：
http://${myApplicationName}/push/{userId}

方案	优点	缺点
广播方案	实现简单	1.增加MQ，可能造成消息挤压、消息顺序的问题 2.每个Websocket Server服务都需要去消费消息，增加每个服务的压力（做无用功） 3.要保证消息幂等性,使用分布式锁会降低服务的并发量
目标询址方案	精确调用Websocket Server服务，性能与并发量更高	1.Id标识为为唯一的服务名：Websocket Server服务名不一致可能导致一些监控系统不便于管控 2.Id标识为唯一的`IP+端口` ：要保证在容器下获取IP的正确性

芋道源码 —— 纯源码解析博客

WebSocket 集群解决方案

前言

方案一：广播方案

代码演示

方案二：目标询址方案（推荐）

代码演示(唯一Id为唯一的服务名的形式)