若依框架整合WebSocket

一、前端

1、定义socket对象

data(){
    return{
        webSocket: null,
    }
}

2、建立连接

//如果webSocket为null进行建立连接
if (!this.webSocket) {
    this.webSocket = new WebSocket(`ws://localhost:8081/handleData/`);
	
    //建立连接
    this.webSocket.onopen = function(event) {
        console.log('WebSocket connected');
    };
	
    //接收消息回调函数
    this.webSocket.onmessage = (event) => {
        // 处理从服务器接收的消息
        console.log(event)
    };
	
    //连接异常，设置webSocket对象为null
    this.webSocket.onerror = function(event) {
        console.error('WebSocket error:', event);
        this.webSocket = null;
    };
	
    //连接关闭，设置webSocket对象为null
    this.webSocket.onclose = function(event) {
        console.log('WebSocket connection closed');
        this.webSocket = null;
    };
}

3、向服务端发送消息

//如果webSocket已经建立连接，则进行数据发送
if (this.webSocket && this.webSocket.readyState === WebSocket.OPEN) {
    const msg = {
		//具体数据
    }
    this.webSocket.send(JSON.stringify(msg));
}

二、后端

1、webSocket配置

@Configuration
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
        return new ServerEndpointExporter();
    }
}

2、webSocket服务建立

@Slf4j
@Component
@ServerEndpoint("/handleData")  //socket连接切入点，即需要进行建立连接的url
@CrossOrigin
public class WebSocketServer {
    
    /** 日志对象 */
    private static Logger logger = LoggerFactory.getLogger(WebSocketServer.class);

    /** 记录当前在线连接数 */
    private static AtomicInteger onlineCount = new AtomicInteger(0);

    /*这是ArrayList的线程安全的版本 读多写少比较适合*/
    private static CopyOnWriteArrayList<Session> sessionList=new CopyOnWriteArrayList<>();

    /**
     * 解决无法注入bean：定义静态service对象，通过@Autowired在系统启动时为静态变量赋值
     * @ Autowired 注解作用在方法上，如果方法没有参数，spring容器会在类加载完后执行一次这个方法，
     * 如果方法中有参数的话，还会从容器中自动注入这个方法的参数，然后执行一次这个方法。
     */
    private static DataService dataService;

    @Autowired
    public void setDeviceDataService(DataService dataService) {
        WebSocketServer.dataService = dataService;
    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) {
        String path = session.getRequestURI().getPath();
        //TODO：在此处获取路径里的传入参数进行建立连接
        
        onlineCount.incrementAndGet();
        sessionList.add(session);
        logger.info("socket连接加入：----deviceId: {}，type: {},当前在线人数为：{}----", deviceId, type, onlineCount.get());
        
		
      	//TODO：为需要做业务处理的dataService增加会话
        MessageDto messageDto = new MessageDto();
        sessionMap.put(session,new SendDataDto(session,messageDto));
		
        //TODO：发送设备数据
        sendMessage(JSONObject.toJSONString(data), session);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        Iterator<Session> iterator = sessionList.iterator();
        while(iterator.hasNext()){
            Session next = iterator.next();
            if(session==next){
                sessionList.remove(session);
                //业务会话集合去除关闭的session
			   sessionMap.remove(session);
                onlineCount.decrementAndGet(); // 在线数减1
                logger.info("连接{}关闭",session);
            }
        }
    }
    /**
    *  连接出现异常调用的方法
    */
    @OnError
    public void one rror(Session session, Throwable error) {
        logger.error("系统发生错误");
	    sessionList.remove(session);
        //业务会话集合去除异常的session
        sessionMap.remove(session);
        error.printStackTrace();
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message
     * 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        logger.info("服务端收到客户端[{}]的消息:{}", session.getId(), message);
        //服务端转发消息给所有的客户端
        MessageDto messageDto = JSONObject.parseObject(message, MessageDto.class);
        //设置节点设备信息
        for (Map.Entry<Session, SendDataDto> entry : sessionMap.entrySet()) {
            if (entry.getKey() == session) {
                MessageDto dto = entry.getValue().getMessageDto();
                //TODO：设置接收到的消息给session的MessageDto
                break;
            }
        }
        //TODO：调用业务处理的service做处理
        
        //发送数据
        sendMessage(JSONObject.toJSONString(data), session);
    }

    /**
     * 服务端发送消息给客户端
     */
    public static synchronized void sendMessage(String message, Session toSession) {
        try {
            logger.info("服务端给客户端[{}]发送消息{}", toSession.getId(), message);
            toSession.getBasicRemote().sendText(message);
        } catch (Exception e) {
            logger.error("服务端发送消息给客户端失败：{}",e.getMessage());
        }
    }
}

3、处理socket消息业务类

@Component
public class DataService {
    
    public static final Logger logger = LoggerFactory.getLogger(DataService.class);

    //需要业务处理的会话集合
    public static ConcurrentHashMap<Session, Object> sessionMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void getAllDevicesList(){
        //TODO：业务处理需要初始化加载的数据
    }

    @Scheduled(fixedRate = 10000)
    public void fetchDataAndSendToFrontend() {
		//TODO：定时给不同的会话发送各自的消息
        for (Map.Entry<Session, Object> entry : sessionMap.entrySet()) {
            logger.info("--------------------------------------------------------------------------------");
            Object value = entry.getValue();
            
            logger.info("--------------------------------------------------------------------------------");
        }
    }

//    @Scheduled(fixedRate = 15000)
    public void getRealTimeData(){
		//业务处理
    }
}