websocket线程优化及发送控制

ruoyi-admin/src/main/java/org/dromara/web/controller/app/EqIotDeviceController.java

@@ -1,49 +1,30 @@
 package org.dromara.web.controller.app;
 
 import cn.dev33.satoken.annotation.SaIgnore;
-import cn.hutool.core.bean.BeanUtil;
-import cn.hutool.core.collection.CollUtil;
-import cn.hutool.http.HttpResponse;
-import cn.hutool.http.HttpUtil;
 import com.alibaba.fastjson.JSONObject;
+import com.baomidou.lock.annotation.Lock4j;
 import com.github.benmanes.caffeine.cache.Cache;
-import jakarta.servlet.http.HttpServletRequest;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
 import org.dromara.common.core.constant.Constants;
-import org.dromara.common.core.domain.R;
 import org.dromara.common.core.domain.model.LoginUser;
-import org.dromara.common.core.exception.ServiceException;
-import org.dromara.common.core.utils.MessageUtils;
-import org.dromara.common.core.utils.StringUtils;
-import org.dromara.common.mybatis.core.page.PageQuery;
-import org.dromara.common.mybatis.core.page.TableDataInfo;
 import org.dromara.common.redis.utils.RedisUtils;
-import org.dromara.common.satoken.utils.LoginHelper;
 import org.dromara.common.web.core.BaseController;
-import org.dromara.common.websocket.dto.WebSocketMessageDto;
+import org.dromara.common.websocket.holder.WebSocketSessionHolder;
-import org.dromara.common.websocket.utils.WebSocketUtils;
-import org.dromara.system.service.ISysDictTypeService;
 import org.dromara.web.config.LastWebSocketParamDto;
 import org.dromara.web.device.iot.DeviceByDeviceName;
 import org.dromara.web.domain.bo.*;
 import org.dromara.web.domain.vo.*;
 import org.dromara.web.service.*;
 import org.dromara.web.utils.WebSocketBusinessHandler;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.util.CollectionUtils;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
 import javax.annotation.Resource;
-import java.text.SimpleDateFormat;
+import java.time.Duration;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.time.ZonedDateTime;
-import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 
 /**
  * app iot设备业务
@@ -51,6 +32,7 @@ import java.util.stream.Collectors;
  * @author Lion Li
  * @date 2024-07-14
  */
+@Log4j2
 @Validated
 @RequiredArgsConstructor
 @RestController
@@ -141,6 +123,7 @@ public class EqIotDeviceController extends BaseController {
      */
     @SaIgnore
     @PostMapping("/sendDeviceMsg2")
+    @Lock4j(keys = {"#eqEquipmentLogBo.equipmentName"}, expire = 30000, acquireTimeout = 1000)
     public void sendDeviceMsg2(@RequestBody EqEquipmentLogBo eqEquipmentLogBo) throws InterruptedException {
 //        Thread.sleep(1000);
         /*String status = eqEquipmentLogBo.getStatus() + "";
@@ -169,12 +152,28 @@ public class EqIotDeviceController extends BaseController {
         Long userId = eqEquipmentVo.getUserId();
         LoginUser loginUser = new LoginUser();
         loginUser.setUserId(userId);
+
+        //小优化 减少不必要查询。 后续可以让设备传设备id 减少iot查询 再通过初始redis来映射userId来减少数据库查询
+        if (!WebSocketSessionHolder.existSession(Long.valueOf("5"+userId.toString().substring(1)))&&!WebSocketSessionHolder.existSession(Long.valueOf("6"+userId.toString().substring(1)))){
+            log.info("app-Socket-{} 未上线",userId);
+            return;
+        }
+
+
         LastWebSocketParamDto dto = RedisUtils.getCacheObject(Constants.WEBSOCKET_METHOD+loginUser.getUserId());
         String param = JSONObject.toJSONString(dto.getBo());
         if (Constants.METHOD_EQ_DETAIL.equals(dto.getMethod())){
-            webSocketBusinessHandler.dealWithDetailMsg(loginUser,param);
+            String redisKey = Constants.METHOD_EQ_DETAIL + userId;
+            Boolean isSent = RedisUtils.setObjectIfAbsent(redisKey, Constants.METHOD_EQ_DETAIL, Duration.ofMillis(1000));
+            if (Boolean.TRUE.equals(isSent)) {
+                webSocketBusinessHandler.dealWithDetailMsg2(loginUser,param,2);
+            }
         }else if (Constants.METHOD_EQ_LIST.equals(dto.getMethod())){
-            webSocketBusinessHandler.dealWithListMsg(loginUser,param);
+            String redisKey = Constants.METHOD_EQ_LIST + userId;
+            Boolean isSent = RedisUtils.setObjectIfAbsent(redisKey, Constants.METHOD_EQ_LIST, Duration.ofMillis(1000));
+            if (Boolean.TRUE.equals(isSent)) {
+                webSocketBusinessHandler.dealWithListMsg2(loginUser,param,2);
+            }
         }else{
             System.out.println("无设备信息最新参数");
         }

ruoyi-admin/src/main/java/org/dromara/web/domain/vo/EqEquipmentListVo.java

@@ -0,0 +1,140 @@
+package org.dromara.web.domain.vo;
+
+import com.alibaba.excel.annotation.ExcelProperty;
+import lombok.Data;
+import org.dromara.common.excel.annotation.ExcelDictFormat;
+import org.dromara.common.excel.convert.ExcelDictConvert;
+
+import java.io.Serial;
+import java.io.Serializable;
+
+
+/**
+ * 设备信息视图对象 eq_equipment
+ *
+ * @author zzc
+ * @date 2024-07-19
+ */
+@Data
+public class EqEquipmentListVo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 主键ID
+     */
+    @ExcelProperty(value = "主键ID")
+    private Long equipmentId;
+
+    /**
+     * 分布id
+     */
+    @ExcelProperty(value = "分布id")
+    private Long areaId;
+
+    /**
+     * 设备类型
+     */
+    @ExcelProperty(value = "设备类型", converter = ExcelDictConvert.class)
+    @ExcelDictFormat(dictType = "eq_equipment_type")
+    private Integer type;
+
+    /**
+     * 登录账号
+     */
+    @ExcelProperty(value = "登录账号ID")
+    private Long userId;
+
+    /**
+     * 设备编码
+     */
+    @ExcelProperty(value = "设备编码")
+    private String equipmentCode;
+
+    /**
+     * 设备名称
+     */
+    @ExcelProperty(value = "设备名称")
+    private String equipmentName;
+
+    /**
+     * 离线状态
+     */
+    @ExcelProperty(value = "离线状态", converter = ExcelDictConvert.class)
+    @ExcelDictFormat(dictType = "eq_line_status")
+    private Integer lineStatus;
+
+    /**
+     * //半开门功能开关：关 false
+     * //半开门功能开关：开 true
+     */
+    private Boolean PartOpenSwitch;
+
+    /**
+     * 排序
+     */
+    @ExcelProperty(value = "排序")
+    private Integer orderNum;
+
+    /**
+     * 设备状态(开合度)
+     * 0.已关闭
+     * 10.打开10%
+     * 20.打开20%
+     * 30.打开30%
+     * 40.打开40%
+     * 50.打开50%
+     * 60.打开60%
+     * 70.打开70%
+     * 80.打开80%
+     * 90.打开90%
+     * 100.打开100%
+     */
+    @ExcelProperty(value = "设备状态 设备的打开状态", converter = ExcelDictConvert.class)
+    @ExcelDictFormat(dictType = "eq_status_type")
+    private Integer openStatus;
+
+    /**
+     * 设备状态(开合度)
+     * 0.已关闭
+     * 10.打开10%
+     * 20.打开20%
+     * 30.打开30%
+     * 40.打开40%
+     * 50.打开50%
+     * 60.打开60%
+     * 70.打开70%
+     * 80.打开80%
+     * 90.打开90%
+     * 100.打开100%
+     * 故障状态
+     */
+    private String openStatusText;
+
+    /**
+     * 设备运行状态
+     * 0.停止中 stop
+     * 1.关闭中 closing
+     * 2.开启中 opening
+     * 11.红外故障 irException
+     * 12.电机运转故障 motorRunException
+     * 13.电机维护故障 motorMaintainException
+     * 14.设备故障 deviceException
+     */
+    @ExcelProperty(value = "设备状态 设备的打开状态", converter = ExcelDictConvert.class)
+    @ExcelDictFormat(dictType = "eq_run_status_type")
+    private Integer runStatus;
+
+    /**
+     * 设备运行状态文本
+     * 0.停止中 stop
+     * 1.关闭中 closing
+     * 2.开启中 openning
+     * 11.红外故障 irException
+     * 12.电机运转故障 motorRunException
+     * 13.电机维护故障 motorMaintainException
+     * 14.设备故障 deviceException
+     */
+    private String runStatusText;
+}

ruoyi-admin/src/main/java/org/dromara/web/utils/WebSocketBusinessHandler.java

@@ -1,6 +1,8 @@
 package org.dromara.web.utils;
 
+import cn.hutool.core.bean.BeanUtil;
 import com.alibaba.fastjson.JSON;
+import com.baomidou.lock.annotation.Lock4j;
 import lombok.extern.slf4j.Slf4j;
 import org.dromara.common.core.constant.Constants;
 import org.dromara.common.core.domain.R;
@@ -18,6 +20,7 @@ import org.dromara.web.controller.app.EqHomeController;
 import org.dromara.web.domain.bo.EqAppHomeBo;
 import org.dromara.web.domain.bo.EqUserConfigBo;
 import org.dromara.web.domain.vo.EqEquipmentDetailVo;
+import org.dromara.web.domain.vo.EqEquipmentListVo;
 import org.dromara.web.domain.vo.EqEquipmentVo;
 import org.dromara.web.domain.vo.EqUserConfigVo;
 import org.dromara.web.service.IEqUserConfigService;
@@ -25,6 +28,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.i18n.LocaleContextHolder;
 import org.springframework.stereotype.Service;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
 
@@ -75,6 +79,42 @@ public class WebSocketBusinessHandler implements WebSocketApi {
             }
         }
 
+//        EqEquipmentDetailVo data = deviceMsg.getData();
+        sendMsg = JsonUtils.toJsonString(deviceMsg);
+        log.info("socket详情发送：{}",sendMsg);
+//        sendMsg(sendMsg, userId);
+        sendMsg(sendMsg, Long.valueOf("5"+userId.toString().substring(1)));
+    }
+
+    @Lock4j(keys = {"#loginUser.userId"}, expire = 2000, acquireTimeout = 1000)
+    @Override
+    public void dealWithDetailMsg2(LoginUser loginUser, String msg,int type) {
+        EqAppHomeBo bo = null;
+        String sendMsg = "";
+        R<EqEquipmentDetailVo> deviceMsg = null;
+        boolean isTrue = true;
+        Long userId = loginUser.getUserId();
+        try {
+            initLanguage(userId);
+            bo = JSON.parseObject(msg, EqAppHomeBo.class);
+            LastWebSocketParamDto dto = new LastWebSocketParamDto();
+            dto.setMethod(Constants.METHOD_EQ_DETAIL);
+            dto.setBo(bo);
+            RedisUtils.setCacheObject(Constants.WEBSOCKET_METHOD+ userId,dto);
+        } catch (Exception e) {
+            log.error("参数错误");
+            isTrue = false;
+            deviceMsg = R.fail("参数错误:"+e.getMessage());
+        }
+        if (isTrue){
+            try {
+                deviceMsg = controller.getDeviceMsg(bo);
+            } catch (Exception e) {
+                deviceMsg = R.fail("设备详情获取失败:"+e.getMessage());
+                log.error("设备详情获取失败");
+            }
+        }
+
 //        EqEquipmentDetailVo data = deviceMsg.getData();
         sendMsg = JsonUtils.toJsonString(deviceMsg);
         log.info("socket详情发送：{}",sendMsg);
@@ -98,7 +138,6 @@ public class WebSocketBusinessHandler implements WebSocketApi {
             }
         }
     }
-
     @Override
     public void dealWithListMsg(LoginUser loginUser, String msg) {
         EqAppHomeBo bo = null;
@@ -137,6 +176,88 @@ public class WebSocketBusinessHandler implements WebSocketApi {
     }
 
 
+    @Lock4j(keys = {"#loginUser.userId"}, expire = 2000, acquireTimeout = 1000)
+    @Override
+    public void dealWithListMsg2(LoginUser loginUser, String msg, int type) {
+        EqAppHomeBo bo = null;
+        String sendMsg = "";
+        boolean isTure = true;
+        Long userId = loginUser.getUserId();
+        try {
+            initLanguage(userId);
+            bo = JSON.parseObject(msg, EqAppHomeBo.class);
+            LastWebSocketParamDto dto = new LastWebSocketParamDto();
+            dto.setMethod(Constants.METHOD_EQ_LIST);
+            dto.setBo(bo);
+            RedisUtils.setCacheObject(Constants.WEBSOCKET_METHOD + userId,dto);
+        } catch (Exception e) {
+            log.error("参数错误");
+            sendMsg = "参数错误";
+            isTure = false;
+        }
+        TableDataInfo<EqEquipmentVo> list = null;
+        boolean canSendFlag = false;
+        if (isTure){
+            PageQuery pageQuery = new PageQuery();
+            pageQuery.setPageNum(0);
+            pageQuery.setPageSize(100);
+            try {
+                list = homeController.listForWebSocket(bo, pageQuery,loginUser);
+                if (type == 1){
+                    canSendFlag = true;
+                }else{
+                    String key = userId + "_" + Constants.METHOD_EQ_LIST;
+                    List<EqEquipmentVo> rows = list.getRows();
+                    List<EqEquipmentListVo> currentVoList = new ArrayList<>();
+                    if (rows !=null && !rows.isEmpty()){
+                        currentVoList = BeanUtil.copyToList(rows, EqEquipmentListVo.class);
+                        if (RedisUtils.hasKey(key)){
+                            List<EqEquipmentVo> cacheList = RedisUtils.getCacheList(key);
+                            for (EqEquipmentVo cacheVo : cacheList) {
+                                for (EqEquipmentListVo currentVo : currentVoList) {
+                                    if (cacheVo.getEquipmentId().equals(currentVo.getEquipmentId())){
+                                        if (
+                                            !cacheVo.getLineStatus().equals(currentVo.getLineStatus())||
+                                                !cacheVo.getAreaId().equals(currentVo.getAreaId())||
+                                                !cacheVo.getOpenStatus().equals(currentVo.getOpenStatus())||
+                                                !cacheVo.getRunStatus().equals(currentVo.getRunStatus())||
+                                                !cacheVo.getPartOpenSwitch().equals(currentVo.getPartOpenSwitch())||
+                                                !cacheVo.getEquipmentCode().equals(currentVo.getEquipmentCode())||
+                                                !cacheVo.getEquipmentName().equals(currentVo.getEquipmentName())||
+                                                !cacheVo.getUserId().equals(currentVo.getUserId())||
+                                                !cacheVo.getOrderNum().equals(currentVo.getOrderNum())
+                                        ){
+                                            canSendFlag = true;
+                                        }
+                                    }
+                                }
+                            }
+
+                        }else{
+                            RedisUtils.setCacheList(key,currentVoList);
+                            canSendFlag = true;
+                        }
+                    }else{
+                        canSendFlag = true;
+                    }
+                }
+            } catch (Exception e) {
+                isTure = false;
+                sendMsg = "获取设备列表失败"+e.getMessage();
+                log.error("获取设备列表失败：{}",e.getMessage());
+            }
+        }
+
+
+        if (canSendFlag){
+            sendMsg = isTure? JsonUtils.toJsonString(list):sendMsg;
+            log.info("socket列表发送：{}",sendMsg);
+            //        sendMsg(sendMsg, userId);
+            sendMsg(sendMsg, Long.valueOf("6"+userId.toString().substring(1)));
+        }
+    }
+
+
     public void sendMsg(String msg,Long userId){
         WebSocketMessageDto dto = new WebSocketMessageDto();
         dto.setMessage(msg);

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/PlusWebSocketHandler.java

@@ -1,8 +1,6 @@
 package org.dromara.common.websocket.handler;
 
 import cn.hutool.core.util.ObjectUtil;
-import cn.hutool.json.JSONObject;
-import cn.hutool.json.JSONUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.dromara.common.core.domain.model.LoginUser;
 import org.dromara.common.websocket.constant.WebSocketConstants;
@@ -10,11 +8,13 @@ import org.dromara.common.websocket.dto.WebSocketMessageDto;
 import org.dromara.common.websocket.holder.WebSocketSessionHolder;
 import org.dromara.common.websocket.utils.WebSocketUtils;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 import org.springframework.web.socket.*;
 import org.springframework.web.socket.handler.AbstractWebSocketHandler;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Objects;
 
 import static org.dromara.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
 
@@ -24,8 +24,10 @@ import static org.dromara.common.websocket.constant.WebSocketConstants.LOGIN_USE
  * @author zendwang
  */
 @Slf4j
+@Service
 public class PlusWebSocketHandler extends AbstractWebSocketHandler {
-
+    @Autowired
+    WebSocketApi webSocketApi;
     /**
      * 连接成功后
      */
@@ -37,7 +39,7 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
             log.info("[connect] invalid token received. sessionId: {}", session.getId());
             return;
         }
-        String currentPath = session.getUri().getPath();
+        String currentPath = Objects.requireNonNull(session.getUri()).getPath();
         Long userId = loginUser.getUserId();
         if (currentPath.contains(WebSocketConstants.detail)){
             WebSocketSessionHolder.addSession(Long.valueOf("5"+userId.toString().substring(1) ), session);
@@ -50,8 +52,7 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
         log.info("[connect] sessionId: {},userId:{},userType:{}", session.getId(), userId, loginUser.getUserType());
     }
 
-    @Autowired
+
-    WebSocketApi webSocketApi;
 
 
     /**
@@ -66,17 +67,17 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
         // 从WebSocket会话中获取登录用户信息
         log.info("socket：接受到消息 {}", message.getPayload());
         LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
-        String currentPath = session.getUri().getPath();
+        String currentPath = Objects.requireNonNull(session.getUri()).getPath();
         /*String lastPath = WebSocketSessionHolder.getSessions(loginUser.getUserId()).getUri().getPath();
         if (!currentPath.equals(lastPath)){
             WebSocketSessionHolder.addSession(loginUser.getUserId(), session);
         }*/
         String payload = message.getPayload();
         if (currentPath.contains(WebSocketConstants.detail)){
-            webSocketApi.dealWithDetailMsg(loginUser, payload);
+            webSocketApi.dealWithDetailMsg2(loginUser, payload,1);
 
         } else if (currentPath.contains(WebSocketConstants.list)) {
-            webSocketApi.dealWithListMsg(loginUser, payload);
+            webSocketApi.dealWithListMsg2(loginUser, payload,1);
 
         } else {
             // 创建WebSocket消息DTO对象

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/WebSocketApi.java

@@ -2,7 +2,6 @@ package org.dromara.common.websocket.handler;
 
 
 import org.dromara.common.core.domain.model.LoginUser;
-import org.springframework.stereotype.Service;
 
 /**
  * @ClassName : WebSocketApi
@@ -10,8 +9,10 @@ import org.springframework.stereotype.Service;
  * @Date : 2024/11/16 19:06
  * @Description :
  **/
-@Service
+//@Service
 public interface WebSocketApi {
     void dealWithDetailMsg(LoginUser loginUser, String msg);
+    void dealWithDetailMsg2(LoginUser loginUser, String msg,int type);
     void dealWithListMsg(LoginUser loginUser, String msg);
+    void dealWithListMsg2(LoginUser loginUser, String msg,int type);
 }