查看原文
其他

WebSocket 集群解决方案

小哈学Java 2024-04-16

来源:juejin.cn/post/7306451559928348709

👉 欢迎加入小哈的星球 ,你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利


全栈前后端分离博客项目 2.0 版本完结啦, 演示链接:http://116.62.199.48/ ,新项目正在酝酿中。全程手摸手,后端 + 前端全栈开发,从 0 到 1 讲解每个功能点开发步骤,1v1 答疑,直到项目上线。目前已更新了239小节,累计38w+字,讲解图:1645张,还在持续爆肝中.. 后续还会上新更多项目,目标是将Java领域典型的项目都整一波,如秒杀系统, 在线商城, IM即时通讯,Spring Cloud Alibaba 等等,戳我加入学习,已有1200+小伙伴加入(早鸟价超低)


前言

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}

图片

👉 欢迎加入小哈的星球 ,你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利


全栈前后端分离博客项目 2.0 版本完结啦, 演示链接:http://116.62.199.48/ ,新项目正在酝酿中。全程手摸手,后端 + 前端全栈开发,从 0 到 1 讲解每个功能点开发步骤,1v1 答疑,直到项目上线。目前已更新了239小节,累计38w+字,讲解图:1645张,还在持续爆肝中.. 后续还会上新更多项目,目标是将Java领域典型的项目都整一波,如秒杀系统, 在线商城, IM即时通讯,Spring Cloud Alibaba 等等,戳我加入学习,已有1200+小伙伴加入(早鸟价超低)



1. 我的私密学习小圈子~

2. 微服务+多级缓存,性能起飞!

3. Nginx静态压缩和代码压缩,提高访问速度!

4. SpringBoot 接口加密解密,新姿势!

最近面试BAT,整理一份面试资料Java面试BATJ通关手册,覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。

获取方式:点“在看”,关注公众号并回复 Java 领取,更多内容陆续奉上。

PS:因公众号平台更改了推送规则,如果不想错过内容,记得读完点一下在看,加个星标,这样每次新文章推送才会第一时间出现在你的订阅列表里。

“在看”支持小哈呀,谢谢啦

继续滑动看下一个
向上滑动看下一个

您可能也对以下帖子感兴趣

文章有问题?点此查看未经处理的缓存