mqtt回调方法调整

zouyu

2025-08-01 7c5e1ef7db84731610d5616b2ee7e6e63abd886d

mqtt回调方法调整

 cnas-device/src/main/java/com/ruoyi/device/controller/WeightController.java

@@ -1,10 +1,14 @@
package com.ruoyi.device.controller;


import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONArray;
import com.ruoyi.common.core.domain.Result;
import com.ruoyi.common.utils.RedisUtil;
import com.ruoyi.device.constant.DCResistanceMqttConstants;
import com.ruoyi.device.dto.WeightRequestDto;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiOperation;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
@@ -12,37 +16,44 @@
import org.springframework.web.bind.annotation.RestController;

import java.util.List;
import java.util.Objects;


@RestController
@RequestMapping("/weight")
public class WeightController {

    @ApiOperation("保存分析天平质量数据到Redis")
    @PostMapping("/handleWeights")
    public ResponseEntity<String> handleWeights(@RequestBody WeightRequestDto weightRequest) {
        try {
            List<Double> weightList = weightRequest.getWeights();
            // åœ¨è¿™é‡Œæ·»åŠ æ•°æ®å¤„理逻辑，例如打印、存储到数据库等
            System.out.println("接收到的数据: " + weightList);
            processValueDataStream(weightList, WeightRequestDto.DENSITY);

            // å¤„理成功返回 200 çŠ¶æ€ç å’Œæ¶ˆæ¯
            return ResponseEntity.ok("数据处理成功");
        } catch (Exception e) {
            // å¤„理失败返回 500 çŠ¶æ€ç å’Œé”™è¯¯æ¶ˆæ¯
            return ResponseEntity.internalServerError().body("数据处理失败: " + e.getMessage());
    public Result<String> handleWeights(@RequestBody WeightRequestDto weightRequest) {
//        try {
//            List<Double> weightList = weightRequest.getWeights();
//            // åœ¨è¿™é‡Œæ·»åŠ æ•°æ®å¤„理逻辑，例如打印、存储到数据库等
//            System.out.println("接收到的数据: " + weightList);
//            processValueDataStream(weightList, WeightRequestDto.DENSITY);
//
//            // å¤„理成功返回 200 çŠ¶æ€ç å’Œæ¶ˆæ¯
//            return ResponseEntity.ok("数据处理成功");
//        } catch (Exception e) {
//            // å¤„理失败返回 500 çŠ¶æ€ç å’Œé”™è¯¯æ¶ˆæ¯
//            return ResponseEntity.internalServerError().body("数据处理失败: " + e.getMessage());
//        }
        if(Objects.nonNull(weightRequest) && CollectionUtil.isNotEmpty(weightRequest.getWeights())){
            RedisUtil.lSet(WeightRequestDto.DENSITY,weightRequest.getWeights());
            return Result.success(null,"保存成功");
        }
        return Result.fail();
    }

    private void processValueDataStream(List<Double> weightList, String dataStream) {
        // å¤„理 weightList ä¸º null çš„情况
        if (weightList == null) {
            // è‹¥ weightList ä¸º null，可选择删除 Redis ä¸­çš„对应键或存入空数组，这里选择存入空数组
            RedisUtil.set(dataStream, "[]");
            return;
        }
        // å°† weightList è½¬æ¢ä¸º JSON å­—符串并存入 Redis
        String jsonStr = JSONArray.toJSONString(weightList);
        RedisUtil.set(dataStream, jsonStr);
    }
//    private void processValueDataStream(List<Double> weightList, String dataStream) {
//        // å¤„理 weightList ä¸º null çš„情况
//        if (weightList == null) {
//            // è‹¥ weightList ä¸º null，可选择删除 Redis ä¸­çš„对应键或存入空数组，这里选择存入空数组
//            RedisUtil.set(dataStream, "[]");
//            return;
//        }
//        // å°† weightList è½¬æ¢ä¸º JSON å­—符串并存入 Redis
//        String jsonStr = JSONArray.toJSONString(weightList);
//        RedisUtil.set(dataStream, jsonStr);
//    }
}

 cnas-device/src/main/java/com/ruoyi/device/service/impl/CollectBridgeServiceImpl.java

@@ -13,7 +13,7 @@
import com.ruoyi.device.mapper.CollectBridgeMapper;
import com.ruoyi.device.pojo.CollectBridge;
import com.ruoyi.device.service.CollectBridgeService;
import com.ruoyi.device.vo.DCResistanceMqttVO;
import com.ruoyi.device.vo.DeviceMqttVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

@@ -111,10 +111,11 @@
    public void addBridgeValueByNS(String jsonStr) {
        Map<String, Object> dataMap = JSONObject.parseObject(jsonStr, new TypeReference<Map<String, Object>>(){}.getType());
        if(CollectionUtil.isNotEmpty(dataMap) && Objects.nonNull(dataMap.get("data"))){
            List<DCResistanceMqttVO> dataList = JSONArray.parseObject(dataMap.get("data").toString(), new TypeReference<List<DCResistanceMqttVO>>(){}.getType());
            List<DeviceMqttVO> dataList = JSONArray.parseObject(dataMap.get("data").toString(), new TypeReference<List<DeviceMqttVO>>(){}.getType());
            //过滤出包含实际电阻值的对象
            // å¤„理 NSDQCS.DQCS.DZZ æ•°æ®
            processSingleValueDataStream(dataList, DCResistanceMqttConstants.NSDQCS_DQCS_DZZ);
            Double value = getValueByDataStream(dataList,DCResistanceMqttConstants.NSDQCS_DQCS_DZZ);
            RedisUtil.set(DCResistanceMqttConstants.NSDQCS_DQCS_DZZ, value);
        }
    }

@@ -122,28 +123,49 @@
    public void addBridgeValueByNSD(String jsonStr) {
        Map<String, Object> dataMap = JSONObject.parseObject(jsonStr, new TypeReference<Map<String, Object>>(){}.getType());
        if(CollectionUtil.isNotEmpty(dataMap) && Objects.nonNull(dataMap.get("data"))){
            List<DCResistanceMqttVO> dataList = JSONArray.parseObject(dataMap.get("data").toString(), new TypeReference<List<DCResistanceMqttVO>>(){}.getType());
            List<DeviceMqttVO> dataList = JSONArray.parseObject(dataMap.get("data").toString(), new TypeReference<List<DeviceMqttVO>>(){}.getType());
            //过滤出包含实际电阻值的对象
            // å¤„理 NSTSSC.TSSC.WY æ•°æ®
            processThreeTimesValueDataStream(dataList, ElongationMqttConstants.NSTSSC_TSSC_WY);
            processMultipleValueDataStream(dataList, ElongationMqttConstants.NSTSSC_TSSC_WY,3);
        }
    }

    /**
     * å¤„理单值数据流的数据并保存到 Redis
     * èŽ·å–mqtt消息体中对应标识的值
     * @param dataList æ•°æ®åˆ—表
     * @param dataStream æ•°æ®æµæ ‡è¯†
     */
    private void processSingleValueDataStream(List<DCResistanceMqttVO> dataList, String dataStream) {
        DCResistanceMqttVO dcResistanceMqttVO = dataList.stream()
    private Double getValueByDataStream(List<DeviceMqttVO> dataList, String dataStream) {
        DeviceMqttVO deviceMqttVO = dataList.stream()
                .filter(f -> StringUtils.equals(f.getDataStream(), dataStream))
                .findFirst()
                .orElse(null);
        if(Objects.nonNull(dcResistanceMqttVO)){
            List<DCResistanceMqttVO.DataPoint> dataPoints = dcResistanceMqttVO.getDataPoints();
        if(Objects.nonNull(deviceMqttVO)){
            List<DeviceMqttVO.DataPoint> dataPoints = deviceMqttVO.getDataPoints();
            BigDecimal value = CollectionUtil.isNotEmpty(dataPoints)?dataPoints.get(0).getValue():BigDecimal.ZERO;
            // ä¿å­˜å•ä¸ªå€¼åˆ° redis
            RedisUtil.set(dataStream, value.doubleValue());
            return value.doubleValue();
        }
        return 0.0;
    }

    /**
     * å¤„理需要保存多个值的数据流，并保存到 Redis
     * @param dataList æ•°æ®åˆ—表
     * @param dataStream æ•°æ®æµæ ‡è¯†
     * @param maxSize ä¿å­˜çš„最大个数
     */
    private void processMultipleValueDataStream(List<DeviceMqttVO> dataList, String dataStream,int maxSize){
        // ç­›é€‰å‡ºå½“前数据流对应的数据
        Double value = getValueByDataStream(dataList, dataStream);
        //判断key是否存在
        boolean existKey = RedisUtil.hasKey(dataStream);
        if(existKey){
            long listSize = RedisUtil.lGetListSize(dataStream);
            if(listSize<maxSize){
                RedisUtil.lSet(dataStream,value);
            }
        }else{
            RedisUtil.lSet(dataStream,value);
        }
    }

@@ -152,7 +174,7 @@
     * @param dataList æ•°æ®åˆ—表
     * @param dataStream æ•°æ®æµæ ‡è¯†
     */
    private void processThreeTimesValueDataStream(List<DCResistanceMqttVO> dataList, String dataStream) {
    private void processThreeTimesValueDataStream(List<DeviceMqttVO> dataList, String dataStream) {
        // ä»Ž Redis èŽ·å–已存储的值
        Object valueFromRedis = RedisUtil.get(dataStream);
        JSONArray valueArray = new JSONArray();
@@ -171,24 +193,7 @@
            }
        }

        // ç­›é€‰å‡ºå½“前数据流对应的数据
        DCResistanceMqttVO dcResistanceMqttVO = dataList.stream()
                .filter(f -> StringUtils.equals(f.getDataStream(), dataStream))
                .findFirst()
                .orElse(null);

        if (Objects.nonNull(dcResistanceMqttVO)) {
            List<DCResistanceMqttVO.DataPoint> dataPoints = dcResistanceMqttVO.getDataPoints();
            if (CollectionUtil.isNotEmpty(dataPoints)) {
                BigDecimal value = dataPoints.get(0).getValue();
                valueArray.add(value.doubleValue());

                // åªä¿ç•™æœ€æ–°çš„三个值
                if (valueArray.size() > 3) {
                    valueArray.remove(0);
                }
            }
        }

        // å°†æ›´æ–°åŽçš„值保存到 Redis
        RedisUtil.set(dataStream, valueArray.toJSONString());

 cnas-device/src/main/java/com/ruoyi/device/vo/DeviceMqttVO.java

ÎļþÃû´Ó cnas-device/src/main/java/com/ruoyi/device/vo/DCResistanceMqttVO.java ÐÞ¸Ä
@@ -7,10 +7,10 @@
import java.util.List;

/**
 * ç›´æµç”µé˜»ï¼šè§£æžmqtt数据响应对象
 * ç½‘口设备(直流电阻、伸长率)：解析mqtt数据响应对象
 */
@Data
public class DCResistanceMqttVO implements Serializable {
public class DeviceMqttVO implements Serializable {

    /**
     * æ•°æ®ç±»åž‹æ ‡è¯†

 ruoyi-common/src/main/java/com/ruoyi/common/utils/RedisUtil.java

@@ -153,6 +153,7 @@

    public static boolean set(String key, Object value) {
        try {

            redisTemplate.opsForValue().set(key, value);
            return true;
        } catch (Exception e) {