区分消息订阅类型，新增上下线监听，修复定时帧缺少CRC32，替换设备在线状态查询方法

Van333

2022-11-16 51af3e4e7c5fb612576b98fd6407f6b65859f374

区分消息订阅类型，新增上下线监听，修复定时帧缺少CRC32，替换设备在线状态查询方法

 ximon-admin/src/main/java/com/sandu/ximon/admin/localMQTT/callback/MsgMqttCallBack.java

ÎļþÃû´Ó ximon-admin/src/main/java/com/sandu/ximon/admin/localMQTT/callback/DefaultMqttCallBack.java ÐÞ¸Ä
@@ -31,8 +31,8 @@
 * @date 2022/11/9 16:24
 */
@Slf4j
@Component("default")
public class DefaultMqttCallBack extends AbsMqttCallBack {
@Component("java_server_msg")
public class MsgMqttCallBack extends AbsMqttCallBack {

    private static final String localMqttConnectTypeOfSync = "1";

@@ -44,7 +44,7 @@
    protected final static ExecutorService EXECUTOR_SERVICE = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50000), new DefaultMqttCallBack.NameTreadFactory());
            new LinkedBlockingQueue<>(50000), new MsgMqttCallBack.NameTreadFactory());

    static class NameTreadFactory implements ThreadFactory {

@@ -60,7 +60,7 @@
    protected void handleReceiveMessage(String topic, String message) {
        EXECUTOR_SERVICE.submit(() -> processMessage(topic,message));

        log.info("接收到消息---DefaultCallBack:topic={},message={}", topic, message);
        log.info("接收到消息---MsgMqttCallBack:topic={},message={}", topic, message);
    }



 ximon-admin/src/main/java/com/sandu/ximon/admin/localMQTT/callback/StatusMqttCallBack.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,90 @@
package com.sandu.ximon.admin.localMQTT.callback;


import com.alibaba.fastjson.JSON;
import com.sandu.ximon.admin.localMQTT.model.LocalMqttMsg;
import com.sandu.ximon.admin.localMQTT.util.HexFrameUtils;
import com.sandu.ximon.admin.manager.iot.amqp.processor.AirDataProcessor;
import com.sandu.ximon.admin.manager.iot.amqp.processor.LightDataProcessor;
import com.sandu.ximon.admin.manager.iot.amqp.processor.PoleMonitorDataProcessor;
import com.sandu.ximon.admin.manager.iot.amqp.processor.c3ChargingProcessor;
import com.sandu.ximon.admin.manager.iot.rrpc.dto.CommonFrame;
import com.sandu.ximon.admin.manager.iot.rrpc.enums.A5OrderEnum;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static com.sandu.ximon.admin.localMQTT.util.MqttClientUtil.MQTT_RETURN_FRAME_MAP;

/**
 * @author van
 * @version 1.0
 * msg：默认回调
 * @date 2022/11/9 16:24
 */
@Slf4j
@Component("java_server_status")
public class StatusMqttCallBack extends AbsMqttCallBack {

    private static final String localMqttConnectStatusConnected = "connected";

    private static final String localMqttConnectStatusDisconnected = "disconnected";

    public static final Map<String, Integer> localMqttConnectStatusMap = new ConcurrentHashMap<>();


    protected final static ExecutorService EXECUTOR_SERVICE = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50000), new StatusMqttCallBack.NameTreadFactory());

    static class NameTreadFactory implements ThreadFactory {

        private final AtomicInteger mThreadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "local-MQTT-msg-thread-" + mThreadNum.getAndIncrement());
        }
    }

    @Override
    protected void handleReceiveMessage(String topic, String message) {
        EXECUTOR_SERVICE.submit(() -> processMessage(topic,message));

        log.info("接收到消息---StatusMqttCallBack:topic={},message={}", topic, message);
    }


    /**
     * åœ¨è¿™é‡Œå¤„理您收到消息后的具体业务逻辑。
     */
    private void processMessage(String topic,String messageString) {
        try {

            String mac = topic.split("/")[4];
            String status = topic.split("/")[5];

            System.out.println("----------------------");
            System.out.println(mac);
            System.out.println(status);
            if (status.equals(localMqttConnectStatusConnected)){
                // è®¾å¤‡æ•°æ®ä¸ŠæŠ¥
                localMqttConnectStatusMap.put(mac,1);
            }
            else if (status.equals(localMqttConnectStatusDisconnected)){
                localMqttConnectStatusMap.put(mac,0);
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}


 ximon-admin/src/main/java/com/sandu/ximon/admin/localMQTT/controller/localMQTTTestController.java

@@ -6,6 +6,7 @@
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import static com.sandu.ximon.admin.localMQTT.callback.StatusMqttCallBack.localMqttConnectStatusMap;
import static java.lang.Thread.sleep;

/**
@@ -21,18 +22,47 @@


    @RequestMapping("/test")
    public static String localMQTT() throws InterruptedException {
    public String localMQTT() throws InterruptedException {

        for (int i = 0; i < 10; i++) {
        for (int i = 0; i < 100; i++) {

            String result1 = MqttClientUtil.sendMqttMsg("363832544E5008FF3A32FFFF",
                    "FEA501000BFE010003FFFF0045971477B6D8C2CA");
            log.info("返回结果:"+result1);
            sleep(1000);
            String result2 = MqttClientUtil.sendMqttMsg("363832544E5008FF3A32FFFF",
                    "FEA501000BFE010003FFFF640F48B1367ABDE0B9");
            log.info("返回结果:"+result2);
            sleep(1000);
            /*
            * å¼€ç¯100
            * FEA501000BFE010003FFFF0045971477B6D8C2CA
            * 10
            * FEA501000BFE010003FFFF0AA542FD69D4E6194E
            * å…³ç¯
            * FEA501000BFE0100030001007130ECA9150640E6
            * æŸ¥è¯¢å¿ƒè·³æ—¶é—´
            * FEA501000AFE110002FFFF26008FBE3DAC7C0D
            * è®¾ç½®å¿ƒè·³30秒
            * FEA501000CFE210004FFFF001E9BB444E9C75BDB49
            * 5分钟
            * FEA501000CFE210004FFFF012C4A7824285825CB53
            * */
//            å¼€10
            String result1 = MqttClientUtil.sendMqttMsg("363832544e5008ff3a32ffff",
                            "FEA501000BFE010003FFFF0AA542FD69D4E6194E");
            log.info("开灯返回结果:"+result1);
            sleep(3000);
//            å…³
            String result2 = MqttClientUtil.sendMqttMsg("363832544e5008ff3a32ffff",
                    "FEA501000BFE0100030001007130ECA9150640E6");
            log.info("关灯返回结果:"+result2);
            sleep(3000);
////            å¿ƒè·³æŸ¥è¯¢
//            String result3 = MqttClientUtil.sendMqttMsg("363832544e5008ff3a32ffff",
//                    "FEA501000AFE110002FFFF26008FBE3DAC7C0D");
//            log.info("心跳查询返回结果:"+result3);
//            sleep(3000);
//            String result4 = MqttClientUtil.sendMqttMsg("363832544e5008ff3a32ffff",
//                    "FEA501001AFE230012FE23000A00017F1019647F111E005428F600EC64EC194EA28A7C");
//            log.info("定时任务返回结果:"+result4);
//            sleep(3000);

//            System.out.println("链接状态：---"+i+"---:");
//            System.out.println(localMqttConnectStatusMap.get("363832544e5008ff3a32ffff"));
//            sleep(10000);

        }


 ximon-admin/src/main/java/com/sandu/ximon/admin/localMQTT/model/LocalMqttMsg.java

@@ -26,6 +26,6 @@
        this.timestamp = String.valueOf(System.currentTimeMillis());
        this.connectType = "1";
        this.msgType = "1";
        this.payload = "FEA501000BFE010003FFFF640F48B1367ABDE0B9";
        this.payload = "";
    }
}

 ximon-admin/src/main/java/com/sandu/ximon/admin/localMQTT/util/MqttClientUtil.java

@@ -32,7 +32,7 @@
    public static String publishPrefix = "v1/devices/request/";


    public static String clientId = "java_server";
    public static String clientId = "java_server_msg";

    public static final Map<String, String> MQTT_RETURN_FRAME_MAP = new ConcurrentHashMap<>();

@@ -63,13 +63,14 @@
                String returnFrame = null ;
                returnFrame = MQTT_RETURN_FRAME_MAP.get(topic);

                for (int i = 0;i < 10;i++){
                    if (!StringUtil.strIsNullOrEmpty(returnFrame)){
                        sleep(500);
                for (int i = 0;i < 50;i++){
                    if (StringUtil.strIsNullOrEmpty(returnFrame)){
                        sleep(100);
                        returnFrame = MQTT_RETURN_FRAME_MAP.get(topic);
                    }else {
                        log.info("返回时间：{} ms",System.currentTimeMillis() - start) ;
                        MQTT_RETURN_FRAME_MAP.remove(topic);
                        String remove = MQTT_RETURN_FRAME_MAP.remove(topic);
                        log.info("remove结果：{} ",remove);
                        return returnFrame;
                    }
                }
@@ -81,6 +82,7 @@

        } catch (MqttException e) {
            log.error("MqttClient send msg faild!", e);
            return("通信超时");
        }
        return null;
    }

 ximon-admin/src/main/java/com/sandu/ximon/admin/manager/iot/amqp/processor/LightDataProcessor.java

@@ -39,9 +39,8 @@
            log.info("心跳相应");
            A5LightHeartbeatReportInnerFrame heartbeatReportInnerFrame = new A5LightHeartbeatReportInnerFrame().transformFrame(frame.getPayload());

            if ("363832544e5008ff1741ffff".equals(deviceName)) {
                System.out.println("心跳包: " + JSON.toJSONString(heartbeatReportInnerFrame));
            }

            if (heartbeatReportInnerFrame.isValidate()) {
                SpringContextHolder.getBean(LightReportDataService.class).saveReportData(deviceName, heartbeatReportInnerFrame.getHeartBeatDataPackage());
                //心跳包上报不保存硬件设备信息

 ximon-admin/src/main/java/com/sandu/ximon/admin/manager/iot/frame/inner/request/A5LightTimerReqInnerFrame.java

@@ -18,9 +18,7 @@
 */
public class A5LightTimerReqInnerFrame implements IRequestInnerFrame {

    private final String payload;
    private final String functionCode = A5LightDataEnum.LightTimer.getCode();
    private final String payloadLength;
    private String payload;

    /**
     * @param framePayload å¤šä¸ªè·¯ç¯å®šæ—¶æŒ‡ä»¤ï¼Œ
@@ -33,12 +31,17 @@
            destinationAddress = lightAddress;
        }
        payload = destinationAddress + framePayload;
        this.payloadLength = SupplementUtils.suppleZero(Integer.toHexString((payload.length() / 2)).toUpperCase(), 4);
//        payload = getEncodeFrame();


    }

    @Override
    public String getEncodeFrame() {
        String functionCode = A5LightDataEnum.LightTimer.getCode();
        String payloadLength = SupplementUtils.suppleZero(Integer.toHexString((payload.length() / 2)).toUpperCase(), 4);
        String frame = functionCode + payloadLength + payload;

        return MQTTConnectTypeEnum.SYNCHRONIZATION.getCode() + frame.toUpperCase() + CRC32Utils.getCRC32(frame.toUpperCase());
    }


 ximon-admin/src/main/java/com/sandu/ximon/admin/manager/iot/rrpc/BaseInvokeSyncService.java

@@ -6,21 +6,26 @@
import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.iot.model.v20180120.*;
import com.sandu.common.execption.BusinessException;
import com.sandu.ximon.admin.dto.DeviceStatus;
import com.sandu.ximon.admin.localMQTT.util.MqttClientUtil;
import com.sandu.ximon.admin.manager.iot.frame.IRequestFrame;
import com.sandu.ximon.admin.manager.iot.frame.inner.BaseResponseInnerFrame;
import com.sandu.ximon.admin.manager.iot.rrpc.dto.CommonFrame;
import com.sandu.ximon.admin.manager.iot.rrpc.dto.InvokeParam;
import com.sandu.ximon.admin.manager.iot.rrpc.dto.WrapResponseCommonFrame;
import com.sandu.ximon.admin.manager.iot.rrpc.enums.DeviceStateEnum;
import com.sandu.ximon.admin.manager.iot.rrpc.topic.IBaseTopic;
import com.sandu.ximon.admin.manager.iot.rrpc.topic.ICustomizeTopic;
import com.sandu.ximon.admin.manager.iot.rrpc.util.FrameUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;

import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.Map;

import static com.sandu.ximon.admin.localMQTT.callback.StatusMqttCallBack.localMqttConnectStatusMap;

/**
 * @author chenjiantian
@@ -54,14 +59,15 @@

    @Override
    public CommonFrame sendRRPC(String deviceName, InvokeParam invokeParam) {
        InvokeThingServiceResponse.Data data = invokeThing(deviceName, invokeParam, false);
        if (data == null) {
            return null;
        }
        String result = data.getResult();
        result = result.replace("\\", "");
        Map map = JSON.parseObject(result, Map.class);
        result = (String) map.get("msg");
//        InvokeThingServiceResponse.Data data = invokeThing(deviceName, invokeParam, false);
//        if (data == null) {
//            return null;
//        }
//        String result = data.getResult();
//        result = result.replace("\\", "");
//        Map map = JSON.parseObject(result, Map.class);
//        result = (String) map.get("msg");
        String result = MqttClientUtil.sendMqttMsg(deviceName,invokeParam.getFrame());
        return FrameUtils.transformMessageToFrame(result);
    }

@@ -80,6 +86,7 @@
//        result = (String) map.get("msg");
        String frame = FrameUtils.transformMessageToFrame(iRequestFrame.getEncodeFrame()).toString();
        String result = MqttClientUtil.sendMqttMsg(deviceName,frame);
        log.info("自定义sendRRPC:请求帧：{},\n,响应帧:{}",iRequestFrame.toString(),result);
        return FrameUtils.transformMessageToFrame(result);
    }

@@ -181,14 +188,23 @@

    @Override
    public List<BatchGetDeviceStateResponse.DeviceStatus> batchGetDeviceState(List<String> deviceNames) {
        BatchGetDeviceStateRequest request = new BatchGetDeviceStateRequest();
        request.setDeviceNames(deviceNames);
        request.setProductKey(getProductKey());
        BatchGetDeviceStateResponse response = invokeSync(request);
        if (response != null && response.getSuccess()) {
            return response.getDeviceStatusList();
        List<BatchGetDeviceStateResponse.DeviceStatus> statusList = new ArrayList<>();

        deviceNames.forEach(l -> {
            BatchGetDeviceStateResponse.DeviceStatus deviceStatus = new BatchGetDeviceStateResponse.DeviceStatus();
                    deviceStatus.setDeviceName(l);

            if (localMqttConnectStatusMap.get(l)!=null &&
                    localMqttConnectStatusMap.get(l)== 1){
                deviceStatus.setStatus("ONLINE");
            }else {
                deviceStatus.setStatus("OFFLINE");
        }
        return null;
            statusList.add(deviceStatus);
        });


        return statusList;
    }

    /**

 ximon-admin/src/main/java/com/sandu/ximon/admin/schedule/C3mOrderSchedule.java

@@ -20,7 +20,7 @@
    private final C3mOrderService c3mOrderService;

    //每10分钟运行一次   åˆ é™¤è¶…时订单
    @Scheduled(cron = "0 0/10 * * * ? ")
//    @Scheduled(cron = "0 0/10 * * * ? ")
    public void deleteOvertimeOrders() {
        c3mOrderService.deleteOrderListByCreateTime();
        log.error("删除超时订单定时任务执行成功");

 ximon-admin/src/main/java/com/sandu/ximon/admin/service/LightService.java

@@ -51,6 +51,8 @@
import java.util.*;
import java.util.stream.Collectors;

import static com.sandu.ximon.admin.localMQTT.callback.StatusMqttCallBack.localMqttConnectStatusMap;

/**
 * @author chenjiantian
 * @date 2021/12/13 16:00
@@ -144,7 +146,8 @@

        // èŽ·å–最近的上报时间
        List<String> deviceCodeList = listLight.stream().map(Light::getDeviceCode).collect(Collectors.toList());
        //拆分list

//        //拆分list
        List<List<String>> split = CollectionUtil.split(deviceCodeList, 100);

        List<BatchGetDeviceStateResponse.DeviceStatus> deviceStatuses = null;
@@ -265,13 +268,17 @@

        List<Map<String, Object>> resultList = new ArrayList<>();
        for (LightControlParam param : paramList) {
            A5LightBrightnessReqInnerFrame lightControlFrame = new A5LightBrightnessReqInnerFrame(param.getBrightness(), param.getLightAddress());
            A5Frame a5Frame = new A5Frame(A5OrderEnum.REQUEST_LIGHT_DATA.getCode(), lightControlFrame);
            A5LightBrightnessReqInnerFrame
                    lightControlFrame = new A5LightBrightnessReqInnerFrame(param.getBrightness(), param.getLightAddress());
            A5Frame a5Frame = new A5Frame(
                    A5OrderEnum.REQUEST_LIGHT_DATA.getCode(),
                    lightControlFrame);
            Map<String, Object> map = new HashMap<>();
            try {
                map.put("deviceCode", param.getDeviceCode());
                WrapResponseCommonFrame<A5LightBrightnessRespInnerFrame> frame
                        = MainBoardInvokeSyncService.getInstance().sendRRPC(param.getDeviceCode(), a5Frame, A5LightBrightnessRespInnerFrame.class);
                        = MainBoardInvokeSyncService.getInstance().sendRRPC
                        (param.getDeviceCode(), a5Frame, A5LightBrightnessRespInnerFrame.class);
                //存储控制帧指令
                StoreOperationRecordsUtils.storeInnerFrameData(param.getDeviceCode(), "单灯帧-亮度控制", a5Frame, frame);


 ximon-admin/src/main/java/com/sandu/ximon/admin/service/LightTaskService.java

@@ -16,6 +16,7 @@
import com.sandu.common.util.SpringContextHolder;
import com.sandu.ximon.admin.dto.LightTaskDto;
import com.sandu.ximon.admin.dto.SingleLightOrderDto;
import com.sandu.ximon.admin.manager.iot.frame.A5Frame;
import com.sandu.ximon.admin.manager.iot.frame.FrameBuilder;
import com.sandu.ximon.admin.manager.iot.frame.IRequestFrame;
import com.sandu.ximon.admin.manager.iot.frame.inner.request.A5LightTimerReqInnerFrame;
@@ -431,10 +432,25 @@
     * @return è¿”回帧
     */
    public A5LightTimerRespInnerFrame sendTimeRRpc(String framePayload, String deviceCode, String lightAddress) {
        IRequestFrame requestFrame = FrameBuilder.builderA5().innerFrame(new A5LightTimerReqInnerFrame(framePayload, lightAddress)).orderType(A5OrderEnum.REQUEST_LIGHT_DATA.getCode()).build();
//        IRequestFrame requestFrame = FrameBuilder.
//                builderA5().
//                innerFrame(
//                        new A5LightTimerReqInnerFrame
//                                (framePayload, lightAddress)
//                                ).
//                orderType(A5OrderEnum.REQUEST_LIGHT_DATA.getCode()).
//                build();

        A5LightTimerReqInnerFrame
                a5LightTimerReqInnerFrame = new A5LightTimerReqInnerFrame(framePayload, lightAddress);
        System.out.println(JSON.toJSONString(a5LightTimerReqInnerFrame) + "          --------a5LightTimerReqInnerFrame");

        A5Frame requestFrame = new A5Frame(A5OrderEnum.REQUEST_LIGHT_DATA.getCode(), a5LightTimerReqInnerFrame);
        System.out.println(requestFrame + "          --------requestFrame");

        WrapResponseCommonFrame<A5LightTimerRespInnerFrame> responseCommonFrame = MainBoardInvokeSyncService.getInstance().sendRRPC(deviceCode, requestFrame, A5LightTimerRespInnerFrame.class);
        WrapResponseCommonFrame<A5LightTimerRespInnerFrame> responseCommonFrame
                = MainBoardInvokeSyncService.getInstance().sendRRPC
                (deviceCode, requestFrame, A5LightTimerRespInnerFrame.class);
        System.out.println(responseCommonFrame + "         -----------responseCommonFrame");
        StoreOperationRecordsUtils.storeInnerFrameData(deviceCode, "单灯帧-控灯", requestFrame, responseCommonFrame);
        return Optional.ofNullable(responseCommonFrame).map(WrapResponseCommonFrame::getResponseInnerFrame).orElse(null);

 ximon-admin/src/main/java/com/sandu/ximon/admin/service/PoleService.java

@@ -60,6 +60,8 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static com.sandu.ximon.admin.localMQTT.callback.StatusMqttCallBack.localMqttConnectStatusMap;

/**
 * ç¯æ†ç›¸å…³
 *
@@ -1064,6 +1066,21 @@
                }
            }
        }
//        List<DeviceStatus> statusList = new ArrayList<>();
//
//        deviceCodeList.forEach(l -> {
//                    DeviceStatus deviceStatus = new DeviceStatus();
//                    deviceStatus.setDeviceCode(l);
//
//            if (localMqttConnectStatusMap.get(l)!=null &&
//                    localMqttConnectStatusMap.get(l)== 1){
//                deviceStatus.setStatus(1);
//            }else {
//                deviceStatus.setStatus(0);
//            }
//            statusList.add(deviceStatus);
//        });

        return statusList;
    }


 ximon-admin/src/main/resources/application-xm.yml

@@ -100,7 +100,7 @@
    broker: tcp://127.0.0.1:1883
    clientList:
      #客户端ID
      - clientId: java_server
      - clientId: java_server_msg
        #监听主题
        subscribeTopic: v1/devices/response/+
        #用户名
@@ -109,3 +109,13 @@
        password: zhxm2512209
        #下发主题
        publishTopic: v1/devices/request/
#监听上下线
      - clientId: java_server_status
        #监听主题
        subscribeTopic: $SYS/brokers/+/clients/#
        #用户名
        userName: server_admin
        #密码
        password: zhxm2512209
        #下发主题
        publishTopic: v1/devices/request/