发现元谋系统农户用水日统计数据中，2025年5月份的数据有重复现象，发生的原因不明，有临时模块中实现删除重复数据的功能。

liurunyu

2025-07-30 d70e143ddc95bb4a422156463484fa5e9896b110

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/largeScreen/WebSocketServer.java

@@ -1,6 +1,11 @@
package com.dy.pipIrrRemote.largeScreen;

import com.dy.common.aop.SsoCheck;
import com.dy.common.aop.SsoVo;
import com.dy.common.springUtil.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import jakarta.websocket.*;
@@ -13,10 +18,13 @@
 * @Author: liurunyu
 * @Date: 2025/2/10 15:25
 * @Description 客户端（浏览器）每建立一个websocket连接，服务端就会创建一个WebSocketServer实例
 * 应用前提是存在一个Config类，如本模块的WebSocketConfig，这时@ServerEndpoint受SpringBoot容器
 * 管理了（在嵌入式web Servlet环境中）
 */
@Slf4j
@ServerEndpoint("/websocket/ls/{id}")
@ServerEndpoint("/websocket/ls/{token}")
@Component
@Scope("prototype") // 非单例，每次请求都会创建新的实例
public class WebSocketServer {

    // 静态变量，记录当前在线连接数
@@ -28,29 +36,46 @@
    // 客户端连接会话，通过它给客户端发送数据
    private Session session;

    // 客户端id
    private String id = "";
    private String orgTag;

    // 客户端上线时刻
    //private String onLineDt ;
    // 客户端
    private String token = "";

    @Autowired
    private SsoCheck ssoCheck ;

    /**
     * 连接建立成功调用的方法
     * @param session websocket会话对象
     * @param id 客户端id
     * @param token 客户端id
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("id") String id) {
        this.session = session;
        this.id = id;
        if(this.id == null || this.id.length() == 0){
            this.id = "" + System.nanoTime() ;
        }
        this.sendMessage(WebSocketHeartBeat.getHeartBeatMessage());
    public void onOpen(Session session, @PathParam("token") String token) {
        if(this.token == null || this.token.length() == 0){
            this.session = session;
            //this.onLineDt = DateTime.yyyy_MM_dd_HH_mm_ss() ;
            this.token = token;
            this.sendMessage(WebSocketHeartBeat.getHeartBeatMessage());

        if (webSocketMap.containsKey(id)) {
            webSocketMap.remove(id);
            webSocketMap.put(id, this);
        } else {
            webSocketMap.put(id, this);
            WebSocketServer.addOnlineCount();
            if (webSocketMap.containsKey(token)) {
                webSocketMap.remove(token);
                webSocketMap.put(token, this);
            } else {
                webSocketMap.put(token, this);
                WebSocketServer.addOnlineCount();
            }
            if(this.ssoCheck == null){
                //2025-07-22 不知为什么，this.ssoCheck会为null
                this.ssoCheck = SpringContextUtil.getBean(SsoCheck.class);
            }
            Object rObj = this.ssoCheck.check(token);
            if(rObj != null) {
                if (rObj instanceof SsoVo ssoVo) {
                    this.orgTag = ssoVo.dataSourceName ;
                }
            }
        }
    }

@@ -59,12 +84,12 @@
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(id)) {
            webSocketMap.remove(id);
        if (webSocketMap.containsKey(token)) {
            webSocketMap.remove(token);
            //从set中删除
            WebSocketServer.subOnlineCount();
        }
        log.info("客户端:" + id + "，关闭了websocket");
        log.info("客户端:" + token + "，关闭了websocket");
    }

    /**
@@ -74,7 +99,7 @@
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("客户端:" + id + "，websocket报文:" + message);
        log.info("客户端:" + token + "，websocket报文:" + message);
    }

    /**
@@ -84,7 +109,7 @@
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("客户端:" + this.id + "，websocket会话异常，原因:" + error.getMessage());
        log.error("客户端:" + this.token + "，websocket会话异常，原因:" + error.getMessage());
    }

    /**
@@ -94,29 +119,35 @@
        try{
            this.session.getBasicRemote().sendText(message);
        }catch (Exception e){
            log.error("客户端:" + id + "，websocket网络发送数据异常", e);
            log.error("客户端:" + token + "，websocket网络发送数据异常", e);
        }
    }

    /**
     * 服务器主动群推送消息
     */
    public static void sendAllMessage(String message) throws IOException {
        ConcurrentHashMap.KeySetView<String, WebSocketServer> ids = webSocketMap.keySet();
        for (String id : ids) {
            WebSocketServer webSocketServer = webSocketMap.get(id);
            webSocketServer.sendMessage(message);
    public static void sendMessage2AllClient(String orgTag, String message) throws IOException {
        ConcurrentHashMap.KeySetView<String, WebSocketServer> tokens = webSocketMap.keySet();
        for (String token : tokens) {
            WebSocketServer webSocketServer = webSocketMap.get(token);
            if(orgTag == null){
                webSocketServer.sendMessage(message);
            }else{
                if(webSocketServer.orgTag != null && webSocketServer.orgTag.equals(orgTag)){
                    webSocketServer.sendMessage(message);
                }
            }
        }
    }

    /**
     * 服务器指定客户端推送消息
     */
    public static void sendOneMessage(String message, String id) throws IOException {
        if (message != null && message.length() != 0 && webSocketMap.containsKey(id)) {
            webSocketMap.get(id).sendMessage(message);
    public static void sendMessage2OneClient(String message, String token) throws IOException {
        if (message != null && message.length() != 0 && webSocketMap.containsKey(token)) {
            webSocketMap.get(token).sendMessage(message);
        } else {
            log.error("客户端" + id + "，不在线！");
            log.error("客户端" + token + "，不在线！");
        }
    }