Websocket集群解决方案( 三 ) _生活百科

ws.onopen调用服务端的@OnOpen注解的方法，储存客户端的session信息，握手建立连接。

客户端调用ws.send发送消息，对应服务端的@OnMessage注解下面的方法接收消息。

服务端调用session.getAsyncRemote().sendText发送消息，对应的客户端ws.onmessage接收消息。

添加 controller@GetMapping({"","index.html"})public ModelAndView index() { ModelAndView view = new ModelAndView("index"); return view;}效果展示打开两个客户端，其中的一个客户端发送消息，另一个客户端也能接收到消息。

文章插图
添加 RabbitMQ 中间件这里使用比较常用的RabbitMQ作为消息中间件，而RabbitMQ支持发布订阅模式：

文章插图
添加消息订阅交换机使用扇形交换机，消息分发给每一条绑定该交换机的队列。以服务器所在的IP + 端口作为唯一标识作为队列的命名，启动一个服务，使用队列绑定交换机，实现消息的订阅：

@Configurationpublic class RabbitConfig {@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("PUBLISH_SUBSCRIBE_EXCHANGE");}@Beanpublic Queue psQueue() throws SocketException {// ip + 端口 为队列名String ip = IpUtils.getServerIp() + "_" + IpUtils.getPort();return new Queue("ps_" + ip);}@Beanpublic Binding routingFirstBinding() throws SocketException {return BindingBuilder.bind(psQueue()).to(fanoutExchange());}}

获取服务器IP和端口可以具体查看Github源码,这里就不做详细描述了。

修改服务端点 ServerEndpoint在WebSocket添加消息的接收方法，@RabbitListener 接收消息，队列名称使用常量命名，动态队列名称使用 #{name}，其中的name是Queue的bean 名称：

@RabbitListener(queues= "#{psQueue.name}")public void pubsubQueueFirst(String message) {System.out.println(message);sendMessage(message);}

然后再调用sendMessage方法发送给所在连接的客户端。
修改消息发送在WebSocket类的onMessage方法将消息发送改成RabbitMQ方式发送:

@OnMessagepublic void onMessage(String message){if (!message.equals("ping")){log.info("【wesocket】收到客户端发送的消息,message={}",message);//sendMessage(message);if (rabbitTemplate == null) {rabbitTemplate = (RabbitTemplate) SpringContextUtil.getBean("rabbitTemplate");}rabbitTemplate.convertAndSend("PUBLISH_SUBSCRIBE_EXCHANGE", null, message);}}

消息通知流程如下所示：