1、完善代码； 2、通信中间件对报警的取水口间隔发送消息改为独立任务完成。

liurunyu

2024-07-31 113e9363391f863d46d5667f8346e3708a6ff3cf

1、完善代码；
2、通信中间件对报警的取水口间隔发送消息改为独立任务完成。

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol/p206V1_0_0/upVos/DataAlarmVo.java

@@ -41,16 +41,16 @@
        StringBuilder str = new StringBuilder() ;
        str.append("      报警:\n");
        str.append("         蓄电池电压：");
        str.append(batteryVolt==null?"":(batteryVolt==1?"报警":"正常"));
        str.append(batteryVolt==null?"":(batteryVolt==1?"报警(1)":"正常(0)"));
        str.append("\n");
        str.append("         漏损：     ");
        str.append(loss==null?"":(loss==1?"报警":"正常"));
        str.append(loss==null?"":(loss==1?"报警(1)":"正常(0)"));
        str.append("\n");
        str.append("         流量计故障：");
        str.append(meter==null?"":(meter==1?"报警":"正常"));
        str.append(meter==null?"":(meter==1?"报警(1)":"正常(0)"));
        str.append("\n");
        str.append("         阀门：     ");
        str.append(valve==null?"":(valve==1?"报警":"正常"));
        str.append(valve==null?"":(valve==1?"报警(1)":"正常(0)"));
        return str.toString() ;
    }
}

 pipIrr-platform/pipIrr-global/src/main/java/com/dy/pipIrrGlobal/config/DingTalk.java

@@ -5,6 +5,7 @@
import com.dingtalk.api.request.OapiRobotSendRequest;
import com.dingtalk.api.response.OapiRobotSendResponse;
import com.taobao.api.ApiException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.binary.Base64;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@@ -23,7 +24,7 @@
 * @LastEditTime 2024-07-31 9:20
 * @Description 钉钉客户端工具类
 */

@Slf4j
@Component
public class DingTalk {
    @Value("${dingtalk.robot.url}")
@@ -72,15 +73,22 @@
            req.setText(text);
            req.setAt(at);
            OapiRobotSendResponse rsp = client.execute(req, CUSTOM_ROBOT_TOKEN);
            System.out.println(rsp.getBody());
            log.info(rsp.getBody());
        } catch (ApiException e) {
            e.printStackTrace();
            log.error("钉钉发送消息异常", e);
            //e.printStackTrace();
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
            log.error("钉钉发送消息异常", e);
            //e.printStackTrace();
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
            log.error("钉钉发送消息异常", e);
            //e.printStackTrace();
        } catch (InvalidKeyException e) {
            throw new RuntimeException(e);
            log.error("钉钉发送消息异常", e);
            //e.printStackTrace();
        } catch (Exception e){
            log.error("钉钉发送消息异常", e);
            //e.printStackTrace();
        }
    }
}

 pipIrr-platform/pipIrr-global/src/main/resources/mapper/RmAlarmStateLastMapper.xml

@@ -340,7 +340,7 @@
      alarm_battery_volt = #{alarmBatteryVolt,jdbcType=TINYINT},
      alarm_valve = #{alarmValve,jdbcType=TINYINT},
      power_type = #{powerType,jdbcType=TINYINT},
      send_ms_time = #{powerType,jdbcType=BIGINT}
      send_ms_time = #{sendMsTime,jdbcType=BIGINT}
    where id = #{id,jdbcType=BIGINT}
  </update>
  <!--根据指定条件获取记录总数-->

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/pom.xml

@@ -125,6 +125,18 @@
            <version>2.0.7</version>
        </dependency>

        <!--钉钉消息推送-->
        <dependency>
            <groupId>com.aliyun</groupId>
            <artifactId>alibaba-dingtalk-service-sdk</artifactId>
            <version>2.0.0</version>
        </dependency>
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
            <version>1.15</version>
        </dependency>

        <!-- 测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/Server.java

@@ -19,6 +19,7 @@
import com.dy.common.mw.protocol.ProtocolUnit;
import com.dy.common.mw.support.SupportUnit;
import com.dy.common.mw.support.SupportUnitConfigVo;
import com.dy.rtuMw.server.tasks.SendMsConstantTask;
import com.dy.rtuMw.server.tasks.ToRtuConstantTask;
import com.dy.rtuMw.resource.ResourceUnit;
import com.dy.rtuMw.resource.ResourceUnitConfigVo;
@@ -340,6 +341,7 @@
            CoreUnit.addConstantTask(new ToRtuConstantTask());
            CoreUnit.addConstantTask(new FromRtuDataConstantTask());
            CoreUnit.addConstantTask(new FromRtuComResultConstantTask());
            CoreUnit.addConstantTask(new SendMsConstantTask());
            coreUnit.start(obj -> {
            });
            units.add(coreUnit) ;

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/forMs/MsObj4Ding.java

New file
@@ -0,0 +1,33 @@
package com.dy.rtuMw.server.forMs;

import com.dy.common.queue.NodeObj;
import com.dy.common.springUtil.SpringContextUtil;
import com.dy.pipIrrGlobal.config.DingTalk;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * @Author: liurunyu
 * @Date: 2024/7/31 18:50
 * @Description
 */
public class MsObj4Ding implements NodeObj {

    private static Logger log = LogManager.getLogger(MsObj4Ding.class.getName());

    public String ms ;

    public MsObj4Ding(String ms){
        this.ms = ms ;
    }

    /**
     * 自己处理自己
     * @return
     */
    public boolean dealSelf(){
        DingTalk dingTalk = SpringContextUtil.getBean(DingTalk.class) ;
        dingTalk.sendMessage(ms);
        return true ;
    }
}

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/forMs/SendMsCache.java

New file
@@ -0,0 +1,68 @@
package com.dy.rtuMw.server.forMs;

import com.dy.common.queue.Node;
import com.dy.common.queue.Queue;
import com.dy.rtuMw.server.ServerProperties;

/**
 * @Author: liurunyu
 * @Date: 2024/7/31 18:47
 * @Description
 */
public class SendMsCache {

    //TCP下行命令缓存队列
    private static Queue cacheQueue = new Queue("SendMsCache") ;

    private static SendMsCache instance = new SendMsCache() ;

    private SendMsCache(){
        cacheQueue.setLimit(ServerProperties.cacheUpDownDataWarnCount, ServerProperties.cacheUpDownDataMaxCount);
    }

    public static SendMsCache getInstance(){
        return instance ;
    }

    /**
     * 缓存命令
     * @param ms
     * @throws Exception
     */
    public static void cacheMs(String ms) throws Exception{
        cacheQueue.pushTail(new MsObj4Ding(ms));
    }

    /**
     * 得到第一个节点
     * @return
     */
    public static Node getFirstQueueNode(){
        return cacheQueue.getFirstNode() ;
    }

    /**
     * 得到最后一个节点
     * @return
     */
    public static Node getLastQueueNode(){
        return cacheQueue.getLastNode() ;
    }

    /**
     * 移除节点
     * @param node
     */
    public static void removeNode(Node node){
        cacheQueue.remove(node);
    }

    /**
     * 缓存的节点数
     * @Return 缓存节点数
     */
    public static Integer size(){
        return cacheQueue.size() ;
    }

}

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/rtuData/p206V1_0_0/TkDealAlarmStatus.java

@@ -6,6 +6,7 @@
import com.dy.pipIrrGlobal.config.DingTalk;
import com.dy.pipIrrGlobal.pojoRm.RmAlarmStateHistory;
import com.dy.rtuMw.server.ServerProperties;
import com.dy.rtuMw.server.forMs.SendMsCache;
import com.dy.rtuMw.server.rtuData.TaskSurpport;
import com.dy.rtuMw.server.rtuData.dbSv.DbSv;
import com.dy.common.mw.protocol.Data;
@@ -170,9 +171,12 @@
     * @param alarmVo
     */
    private void sendMessage(PrController controller, DataAlarmVo alarmVo){
        String ms = "取水口" + controller.getIntakeName() + "产生报警：" + alarmVo.alarmContent() ;
        DingTalk dingTalk = SpringContextUtil.getBean(DingTalk.class) ;
        dingTalk.sendMessage(ms);
        String ms = "取水口“" + controller.getIntakeName() + "”产生报警：" + alarmVo.alarmContent() ;
        try{
            SendMsCache.cacheMs(ms);
        }catch (Exception e){
            log.error("缓存钉钉消息异常：", e);
        }
    }

}

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/tasks/SendMsConstantTask.java

New file
@@ -0,0 +1,81 @@
package com.dy.rtuMw.server.tasks;

import com.dy.common.mw.core.CoreTask;
import com.dy.common.queue.Node;
import com.dy.rtuMw.server.forMs.MsObj4Ding;
import com.dy.rtuMw.server.forMs.SendMsCache;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * @Author: liurunyu
 * @Date: 2024/7/31 18:45
 * @Description
 */
public class SendMsConstantTask  extends CoreTask {

    private static final Logger log = LogManager.getLogger(SendMsConstantTask.class.getName());

    /**
     * 在单线程环境中运行
     */
    @Override
    public Integer execute() {
        try{
            //log.info("当前消息队列中还有结点数量：" + SendMsCache.size());
            dealMs() ;
        }catch(Exception e){
            log.error(e);
        }
        return SendMsCache.size()>0?0:1 ;
    }

    /**
     * 处理下行命令
     */
    public void dealMs() {
        Node first = SendMsCache.getFirstQueueNode() ;
        if(first != null){
            Node last = SendMsCache.getLastQueueNode() ;
            while (last != null){
                last = this.doDealMs(first, last);
            }
        }
    }
    /**
     * 处理缓存的下行节点
     * @param first 第一个节点
     * @param last 最后一个节点
     */
    private Node doDealMs(Node first, Node last){
        if(last != null){
            //在dealNode方法中，可能要把last从队列中移除，这时last.pre为空，所以提前把last.pre取出来
            Node pre = last.pre ;
            dealNode(last) ;
            if(first != last){
                return pre ;
            }else{
                //停止
                return null ;
            }
        }else{
            return null ;
        }
    }

    /**
     * 处理一个节点
     * @param node 节点
     */
    private void dealNode(Node node){
        if(node != null && node.obj != null){
            MsObj4Ding obj = (MsObj4Ding)node.obj ;
            boolean removeNode = obj.dealSelf() ;
            if(removeNode){
                SendMsCache.removeNode(node);
            }
        }
    }


}

 pipIrr-platform/pipIrr-web/pipIrr-mwTest-web/src/main/java/com/dy/pipIrrMwTestWeb/p206V1_0_0/CommandP206V1_0_0Ctrl.java

@@ -242,7 +242,7 @@
        comVo.moneyRemain = 234.56 ;
        comVo.waterPrice = 1.2 ;
        comVo.minutes = 3 ;
        String nextDtStr = DateTime.nextXMinute_yyyy_MM_dd_HH_mm_ss(DateTime.yyyy_MM_dd_HH_mm_ss(), 5) ;//将来5分钟
        String nextDtStr = DateTime.nextXMinute_yyyy_MM_dd_HH_mm_ss(DateTime.yyyy_MM_dd_HH_mm_ss(), comVo.minutes) ;//将来X分钟
        int[] nextDt = DateTime.yyyy_MM_dd_HH_MM_SS_2_ymdhmsGroup(nextDtStr) ;
        comVo.year = nextDt[0] ;
        comVo.month = nextDt[1] ;