最近刚好有个场景，业务处理一份报告需要关注实时处理的进度。
本来打算使用前端轮训方式，但是考虑到这样效率比较低，也无法精确知道处理进度，就想到用websocket和前端实时交互，进度有更新就通知前端，避免了无用的空轮训请求。
websocket通过session链接，和前端保持链接，将客户的信息存储在内存中，用户每个请求都有一个唯一的uuid,后续通过uuid查找对应客户的session发送websocket。
系统架构图
流程图
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** @author * @since 2024/8/26*/
@Configuration
@EnableAutoConfiguration
@EnableWebSocket
public class WebSocketConfig {@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}}
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;/*** @author * @since 2024/8/26*/
@ServerEndpoint(value = "/websocket/progress")
@Component
@Slf4j
public class WebSocketServer {//静态变量，记录当前在线连接数private static int onlineCount = 0;//存储客户端对应的WebSocket对象private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet();//与某个客户端的连接会话private Session session;//客户端sidprivate String sid = "";//建立连接成功后调用的方法@OnOpenpublic void onOpen(Session session) {//获取当前会话this.session = session;//存储WebSocket对象String sid = getUuid();//客户端标识this.sid = sid;webSocketSet.add(this);//在线连接计数addOnlineCount();try {//调用服务端发送信息方法sendMessage("连接成功" + this.sid);} catch (Exception e) {log.error("IO异常：{} ", e);}}private String getUuid() {IdWorker idWorker = new IdWorker();long l = idWorker.nextId();return String.valueOf(l);}@OnClosepublic void onClose() {//移除WebSocket对象webSocketSet.remove(this);subOnlineCount();}@OnMessagepublic void onMessage(String message) {log.info("WebSocketServer 收到message{}", this.sid);}@OnErrorpublic void onError(Throwable error) {log.error("onError,{}", error);}public void sendMessage(String message) throws IOException {this.session.getBasicRemote().sendText(message);}public static void sendInfo(String message, String sid) {if (sid == null) {return;}for (WebSocketServer item : webSocketSet) {try {if (item.sid.equals(sid)) {//发给单一用户item.sendMessage(message);}} catch (IOException e) {log.error("发送websocket失败{}", e);}}}public static synchronized int getOnlineCount() {return onlineCount;}public static synchronized void addOnlineCount() {WebSocketServer.onlineCount += 1;}public static synchronized void subOnlineCount() {WebSocketServer.onlineCount -= 1;}
}
和前端本地测试的好好的，一点毛病没有，以为万事大吉了。
结果一发布到测试环境，gg了，进度条时好时坏，一开始还没查出来问题，没有往集群的方面想，还以为是哪里不小心自动关闭了websocket，查了n久之后，突然顿悟了，有集群啊，多台服务器，前端链接之后，session是保持了，但是一开始链接的可能是服务器A，后续的请求可能就到服务器B了，那就拿不到客户的session，就无法发送消息了。
找到问题，就得想解决方法了
方法一 固定ip请求，用nginx的ip_hash负载均衡请求，可惜行不通，问了运维那边，ip是动态了，搞不定。
方法二 改网关的负载均衡策略，因为原来的负载均衡策略是均衡分发，改到源地址策略，就会固定ip请求到同一服务器了，和nginx ip_hash类似，但是也行不通，不能改策略。
方法三 最后只剩这个方法，通过rabbitmq广播各个服务器，根据服务器本身存储的session信息(uuid)，判断这个消息时候需要处理。
customer自己发送消息，自己监听，发送mq的时候会自动在头尾带上"，以及会对字符串中的“引号进行转义，加上\，所以都要进行替换
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;import java.io.IOException;/*** @author * @since 2024/9/19*/
@Component
public class WebsocketMqCustomer {@RabbitListener(queues = "#{psQueue.name}")public void pubsubMqMsg(String message) throws IOException {String[] xxxx = message.replace("\\", "").split(";;;");WebSocketServer.sendInfo(xxxx[1].substring(0, xxxx[1].length() - 1).replace("\\", ""), xxxx[0].substring(1));}public static void sendInfo(String message, String sid) {if (StringUtils.isNotEmpty(sid)) {RabbitTemplate rabbitTemplate = SpringUtil.getBean(RabbitTemplate.class);rabbitTemplate.convertAndSend(MainRabbitMqCreateConfig.EXCHANGE_WEBSOCKET, sid, sid + ";;;" + message);}}
}
还遇见了一个问题，@ServerEndpoint注解添加的类无法通过@Autowired直接注入，因为websocket处理是基于Servlet的，而Servlet容器并不像Spring mvc那样自动管理bean的依赖注入，可以通过Application来获取其他bean。
