Merge branch 'wvp-28181-2.0' into Zafu-Dev-1127

648540858

2022-11-29 ef6693aabbbf12e83d09ad8749f6e60faacc012d

Merge branch 'wvp-28181-2.0' into Zafu-Dev-1127

 sql/mysql.sql

@@ -1,4 +1,4 @@
-- MySQL dump 10.13  Distrib 8.0.30, for Linux (x86_64)
-- MySQL dump 10.13  Distrib 8.0.31, for Linux (x86_64)
--
-- Host: 127.0.0.1    Database: wvp
-- ------------------------------------------------------
@@ -34,13 +34,13 @@
                          `online` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `registerTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `keepaliveTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `ip` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `ip` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                          `updateTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                          `port` int DEFAULT NULL,
                          `expires` int DEFAULT NULL,
                          `subscribeCycleForCatalog` int DEFAULT NULL,
                          `hostAddress` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `hostAddress` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `charset` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                          `subscribeCycleForMobilePosition` int DEFAULT NULL,
                          `mobilePositionSubmissionInterval` int DEFAULT '5',
@@ -48,12 +48,13 @@
                          `ssrcCheck` int DEFAULT '0',
                          `geoCoordSys` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                          `treeType` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                          `mediaServerId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT 'auto',
                          `custom_name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `password` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `custom_name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `password` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `sdpIp` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL,
                          `localIp` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL,
                          PRIMARY KEY (`id`),
                          UNIQUE KEY `device_deviceId_uindex` (`deviceId`)
) ENGINE=InnoDB AUTO_INCREMENT=47 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
) ENGINE=InnoDB AUTO_INCREMENT=57 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -145,7 +146,7 @@
                                  PRIMARY KEY (`id`),
                                  UNIQUE KEY `device_channel_id_uindex` (`id`),
                                  UNIQUE KEY `device_channel_pk` (`channelId`,`deviceId`)
) ENGINE=InnoDB AUTO_INCREMENT=60301 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
) ENGINE=InnoDB AUTO_INCREMENT=74416 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -215,7 +216,7 @@
                             PRIMARY KEY (`gbStreamId`) USING BTREE,
                             UNIQUE KEY `app` (`app`,`stream`) USING BTREE,
                             UNIQUE KEY `gbId` (`gbId`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=301059 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
) ENGINE=InnoDB AUTO_INCREMENT=331060 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -245,7 +246,7 @@
                       `username` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                       `createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                       PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=733627 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
) ENGINE=InnoDB AUTO_INCREMENT=760908 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -337,7 +338,7 @@
                                   PRIMARY KEY (`id`),
                                   UNIQUE KEY `parent_platform_id_uindex` (`id`),
                                   UNIQUE KEY `parent_platform_pk` (`serverGBId`)
) ENGINE=InnoDB AUTO_INCREMENT=44 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
) ENGINE=InnoDB AUTO_INCREMENT=47 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -389,7 +390,7 @@
                                       `catalogId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                                       `deviceChannelId` int NOT NULL,
                                       PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=102 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
) ENGINE=InnoDB AUTO_INCREMENT=3146 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -415,7 +416,7 @@
                                      `id` int NOT NULL AUTO_INCREMENT,
                                      PRIMARY KEY (`id`),
                                      UNIQUE KEY `platform_gb_stream_pk` (`platformId`,`catalogId`,`gbStreamId`)
) ENGINE=InnoDB AUTO_INCREMENT=301766 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
) ENGINE=InnoDB AUTO_INCREMENT=391772 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -457,7 +458,7 @@
                                `enable_disable_none_reader` bit(1) DEFAULT NULL,
                                PRIMARY KEY (`id`),
                                UNIQUE KEY `stream_proxy_pk` (`app`,`stream`)
) ENGINE=InnoDB AUTO_INCREMENT=548 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
) ENGINE=InnoDB AUTO_INCREMENT=568 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -494,7 +495,7 @@
                               `self` int DEFAULT NULL,
                               PRIMARY KEY (`id`),
                               UNIQUE KEY `stream_push_pk` (`app`,`stream`)
) ENGINE=InnoDB AUTO_INCREMENT=310558 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
) ENGINE=InnoDB AUTO_INCREMENT=361492 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;

--
@@ -572,4 +573,4 @@
/*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */;
/*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;

-- Dump completed on 2022-10-18 17:00:02
-- Dump completed on 2022-11-29 11:47:46

 src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java

@@ -29,8 +29,6 @@
    @Autowired

    private ISIPProcessorObserver sipProcessorObserver;







    private final Map<String, SipProviderImpl> tcpSipProviderMap = new ConcurrentHashMap<>();

    private final Map<String, SipProviderImpl> udpSipProviderMap = new ConcurrentHashMap<>();




 src/main/java/com/genersoft/iot/vmp/gb28181/auth/DigestServerAuthenticationHelper.java

@@ -25,10 +25,9 @@
 */
package com.genersoft.iot.vmp.gb28181.auth;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.util.Random;
import gov.nist.core.InternalErrorHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sip.address.URI;
import javax.sip.header.AuthorizationHeader;
@@ -36,10 +35,10 @@
import javax.sip.header.WWWAuthenticateHeader;
import javax.sip.message.Request;
import javax.sip.message.Response;

import gov.nist.core.InternalErrorHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.util.Random;

/**
 * Implements the HTTP digest authentication method server side functionality.
@@ -201,12 +200,13 @@
        // String ncStr = new DecimalFormat("00000000").format(Integer.parseInt(nc + "", 16));

        String A1 = username + ":" + realm + ":" + pass;

        String A2 = request.getMethod().toUpperCase() + ":" + uri.toString();

        byte mdbytes[] = messageDigest.digest(A1.getBytes());
        String HA1 = toHexString(mdbytes);
        logger.debug("A1: " + A1);
        logger.debug("A2: " + A2);

        mdbytes = messageDigest.digest(A2.getBytes());
        String HA2 = toHexString(mdbytes);
        logger.debug("HA1: " + HA1);
@@ -238,58 +238,4 @@

    }

//     public static void main(String[] args) throws NoSuchAlgorithmException {
//         String realm = "3402000000";
//         String username = "44010000001180008012";


//         String nonce = "07cab60999fbf643264ace27d3b7de8b";
//         String uri = "sip:34020000002000000001@3402000000";
//         // qop 保护质量 包含auth（默认的）和auth-int（增加了报文完整性检测）两种策略
//         String qop = "auth";

//         // 客户端随机数，这是一个不透明的字符串值，由客户端提供，并且客户端和服务器都会使用，以避免用明文文本。
//         // 这使得双方都可以查验对方的身份，并对消息的完整性提供一些保护
//         //String cNonce = authHeader.getCNonce();

//         // nonce计数器，是一个16进制的数值，表示同一nonce下客户端发送出请求的数量
//         int nc = 1;
//         String ncStr = new DecimalFormat("00000000").format(nc);
// //        String ncStr = new DecimalFormat("00000000").format(Integer.parseInt(nc + "", 16));
//         MessageDigest messageDigest = MessageDigest.getInstance(DEFAULT_ALGORITHM);
//         String A1 = username + ":" + realm + ":" + "12345678";
//         String A2 = "REGISTER" + ":" + uri;
//         byte mdbytes[] = messageDigest.digest(A1.getBytes());
//         String HA1 = toHexString(mdbytes);
//         System.out.println("A1: " + A1);
//         System.out.println("A2: " + A2);

//         mdbytes = messageDigest.digest(A2.getBytes());
//         String HA2 = toHexString(mdbytes);
//         System.out.println("HA1: " + HA1);
//         System.out.println("HA2: " + HA2);
//         String cnonce = "0a4f113b";
//         System.out.println("nonce: " + nonce);
//         System.out.println("nc: " + ncStr);
//         System.out.println("cnonce: " + cnonce);
//         System.out.println("qop: " + qop);
//         String KD = HA1 + ":" + nonce;

//         if (qop != null && qop.equals("auth") ) {
//             if (nc != -1) {
//                 KD += ":" + ncStr;
//             }
//             if (cnonce != null) {
//                 KD += ":" + cnonce;
//             }
//             KD += ":" + qop;
//         }
//         KD += ":" + HA2;
//         System.out.println("KD: " + KD);
//         mdbytes = messageDigest.digest(KD.getBytes());
//         String mdString = toHexString(mdbytes);
//         System.out.println("mdString: " + mdString);
//         String response = "4f0507d4b87cdecff04bdaf4c96348f0";
//         System.out.println("response: " + response);
//     }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.gb28181.SipLayer;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
@@ -9,13 +8,13 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import com.genersoft.iot.vmp.utils.DateUtil;
import gov.nist.javax.sip.message.MessageFactoryImpl;
import gov.nist.javax.sip.message.SIPRequest;
import org.slf4j.Logger;
@@ -26,8 +25,10 @@
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.sip.*;
import javax.sip.header.*;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import javax.sip.header.CallIdHeader;
import javax.sip.header.WWWAuthenticateHeader;
import javax.sip.message.Request;
import java.text.ParseException;
import java.util.ArrayList;

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java

@@ -11,7 +11,6 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
@@ -31,7 +30,6 @@
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
@@ -77,8 +75,6 @@
    @Autowired
    private IDeviceChannelService deviceChannelService;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
@@ -98,9 +94,9 @@
        }catch (SipException | InvalidArgumentException | ParseException e) {
            e.printStackTrace();
        }
        boolean runed = !taskQueue.isEmpty();
        taskQueue.offer(new HandlerCatchData(evt, null, null));
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (!runed) {
            taskExecutor.execute(()-> {
                while (!taskQueue.isEmpty()) {
                    try {
@@ -128,7 +124,6 @@
                        logger.error("处理NOTIFY消息时错误", e);
                    }
                }
                taskQueueHandlerRun = false;
            });
        }
    }

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java

@@ -9,7 +9,6 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.NotifyMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IDeviceAlarmService;
@@ -27,17 +26,15 @@
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.message.Response;

import java.text.ParseException;
import java.util.concurrent.ConcurrentLinkedQueue;

import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.*;
import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText;

/**
 * 报警事件的处理，参考：9.4
@@ -72,8 +69,6 @@
    @Autowired
    private IDeviceChannelService deviceChannelService;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<SipMsgInfo> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
@@ -89,128 +84,128 @@
    @Override
    public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
        logger.info("[收到报警通知]设备：{}", device.getDeviceId());

        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(new SipMsgInfo(evt, device, rootElement));
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        // 回复200 OK
        try {
            responseAck((SIPRequest) evt.getRequest(), Response.OK);
        } catch (SipException | InvalidArgumentException | ParseException e) {
            logger.error("[命令发送失败] 报警通知回复: {}", e.getMessage());
        }
        if (isEmpty) {
            taskExecutor.execute(() -> {
                logger.info("[处理报警通知]待处理数量：{}", taskQueue.size() );
                while (!taskQueue.isEmpty()) {
                    SipMsgInfo sipMsgInfo = taskQueue.poll();
                    // 回复200 OK
                    try {
                        responseAck((SIPRequest) sipMsgInfo.getEvt().getRequest(), Response.OK);
                    } catch (SipException | InvalidArgumentException | ParseException e) {
                        logger.error("[处理报警通知], 回复200OK失败", e);
                    }
                        SipMsgInfo sipMsgInfo = taskQueue.poll();

                    Element deviceIdElement = sipMsgInfo.getRootElement().element("DeviceID");
                    String channelId = deviceIdElement.getText().toString();
                        Element deviceIdElement = sipMsgInfo.getRootElement().element("DeviceID");
                        String channelId = deviceIdElement.getText().toString();

                    DeviceAlarm deviceAlarm = new DeviceAlarm();
                    deviceAlarm.setCreateTime(DateUtil.getNow());
                    deviceAlarm.setDeviceId(sipMsgInfo.getDevice().getDeviceId());
                    deviceAlarm.setChannelId(channelId);
                    deviceAlarm.setAlarmPriority(getText(sipMsgInfo.getRootElement(), "AlarmPriority"));
                    deviceAlarm.setAlarmMethod(getText(sipMsgInfo.getRootElement(), "AlarmMethod"));
                    String alarmTime = XmlUtil.getText(sipMsgInfo.getRootElement(), "AlarmTime");
                    if (alarmTime == null) {
                        continue;
                    }
                    deviceAlarm.setAlarmTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(alarmTime));
                    String alarmDescription = getText(sipMsgInfo.getRootElement(), "AlarmDescription");
                    if (alarmDescription == null) {
                        deviceAlarm.setAlarmDescription("");
                    } else {
                        deviceAlarm.setAlarmDescription(alarmDescription);
                    }
                    String longitude = getText(sipMsgInfo.getRootElement(), "Longitude");
                    if (longitude != null && NumericUtil.isDouble(longitude)) {
                        deviceAlarm.setLongitude(Double.parseDouble(longitude));
                    } else {
                        deviceAlarm.setLongitude(0.00);
                    }
                    String latitude = getText(sipMsgInfo.getRootElement(), "Latitude");
                    if (latitude != null && NumericUtil.isDouble(latitude)) {
                        deviceAlarm.setLatitude(Double.parseDouble(latitude));
                    } else {
                        deviceAlarm.setLatitude(0.00);
                    }
                        DeviceAlarm deviceAlarm = new DeviceAlarm();
                        deviceAlarm.setCreateTime(DateUtil.getNow());
                        deviceAlarm.setDeviceId(sipMsgInfo.getDevice().getDeviceId());
                        deviceAlarm.setChannelId(channelId);
                        deviceAlarm.setAlarmPriority(getText(sipMsgInfo.getRootElement(), "AlarmPriority"));
                        deviceAlarm.setAlarmMethod(getText(sipMsgInfo.getRootElement(), "AlarmMethod"));
                        String alarmTime = XmlUtil.getText(sipMsgInfo.getRootElement(), "AlarmTime");
                        if (alarmTime == null) {
                            continue;
                        }
                        deviceAlarm.setAlarmTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(alarmTime));
                        String alarmDescription = getText(sipMsgInfo.getRootElement(), "AlarmDescription");
                        if (alarmDescription == null) {
                            deviceAlarm.setAlarmDescription("");
                        } else {
                            deviceAlarm.setAlarmDescription(alarmDescription);
                        }
                        String longitude = getText(sipMsgInfo.getRootElement(), "Longitude");
                        if (longitude != null && NumericUtil.isDouble(longitude)) {
                            deviceAlarm.setLongitude(Double.parseDouble(longitude));
                        } else {
                            deviceAlarm.setLongitude(0.00);
                        }
                        String latitude = getText(sipMsgInfo.getRootElement(), "Latitude");
                        if (latitude != null && NumericUtil.isDouble(latitude)) {
                            deviceAlarm.setLatitude(Double.parseDouble(latitude));
                        } else {
                            deviceAlarm.setLatitude(0.00);
                        }

                    if (!ObjectUtils.isEmpty(deviceAlarm.getAlarmMethod())) {
                        if ( deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.GPS.getVal() + "")) {
                            MobilePosition mobilePosition = new MobilePosition();
                            mobilePosition.setCreateTime(DateUtil.getNow());
                            mobilePosition.setDeviceId(deviceAlarm.getDeviceId());
                            mobilePosition.setTime(deviceAlarm.getAlarmTime());
                            mobilePosition.setLongitude(deviceAlarm.getLongitude());
                            mobilePosition.setLatitude(deviceAlarm.getLatitude());
                            mobilePosition.setReportSource("GPS Alarm");
                        if (!ObjectUtils.isEmpty(deviceAlarm.getAlarmMethod())) {
                            if ( deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.GPS.getVal() + "")) {
                                MobilePosition mobilePosition = new MobilePosition();
                                mobilePosition.setCreateTime(DateUtil.getNow());
                                mobilePosition.setDeviceId(deviceAlarm.getDeviceId());
                                mobilePosition.setTime(deviceAlarm.getAlarmTime());
                                mobilePosition.setLongitude(deviceAlarm.getLongitude());
                                mobilePosition.setLatitude(deviceAlarm.getLatitude());
                                mobilePosition.setReportSource("GPS Alarm");

                            // 更新device channel 的经纬度
                            DeviceChannel deviceChannel = new DeviceChannel();
                            deviceChannel.setDeviceId(sipMsgInfo.getDevice().getDeviceId());
                            deviceChannel.setChannelId(channelId);
                            deviceChannel.setLongitude(mobilePosition.getLongitude());
                            deviceChannel.setLatitude(mobilePosition.getLatitude());
                            deviceChannel.setGpsTime(mobilePosition.getTime());
                                // 更新device channel 的经纬度
                                DeviceChannel deviceChannel = new DeviceChannel();
                                deviceChannel.setDeviceId(sipMsgInfo.getDevice().getDeviceId());
                                deviceChannel.setChannelId(channelId);
                                deviceChannel.setLongitude(mobilePosition.getLongitude());
                                deviceChannel.setLatitude(mobilePosition.getLatitude());
                                deviceChannel.setGpsTime(mobilePosition.getTime());

                            deviceChannel = deviceChannelService.updateGps(deviceChannel, sipMsgInfo.getDevice());
                                deviceChannel = deviceChannelService.updateGps(deviceChannel, sipMsgInfo.getDevice());

                            mobilePosition.setLongitudeWgs84(deviceChannel.getLongitudeWgs84());
                            mobilePosition.setLatitudeWgs84(deviceChannel.getLatitudeWgs84());
                            mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
                            mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());
                                mobilePosition.setLongitudeWgs84(deviceChannel.getLongitudeWgs84());
                                mobilePosition.setLatitudeWgs84(deviceChannel.getLatitudeWgs84());
                                mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
                                mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());

                            if (userSetting.getSavePositionHistory()) {
                                storager.insertMobilePosition(mobilePosition);
                                if (userSetting.getSavePositionHistory()) {
                                    storager.insertMobilePosition(mobilePosition);
                                }
                                storager.updateChannelPosition(deviceChannel);

                                // 发送redis消息。 通知位置信息的变化
                                JSONObject jsonObject = new JSONObject();
                                jsonObject.put("time", mobilePosition.getTime());
                                jsonObject.put("serial", deviceChannel.getDeviceId());
                                jsonObject.put("code", deviceChannel.getChannelId());
                                jsonObject.put("longitude", mobilePosition.getLongitude());
                                jsonObject.put("latitude", mobilePosition.getLatitude());
                                jsonObject.put("altitude", mobilePosition.getAltitude());
                                jsonObject.put("direction", mobilePosition.getDirection());
                                jsonObject.put("speed", mobilePosition.getSpeed());
                                redisCatchStorage.sendMobilePositionMsg(jsonObject);
                            }
                            storager.updateChannelPosition(deviceChannel);

                            // 发送redis消息。 通知位置信息的变化
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put("time", mobilePosition.getTime());
                            jsonObject.put("serial", deviceChannel.getDeviceId());
                            jsonObject.put("code", deviceChannel.getChannelId());
                            jsonObject.put("longitude", mobilePosition.getLongitude());
                            jsonObject.put("latitude", mobilePosition.getLatitude());
                            jsonObject.put("altitude", mobilePosition.getAltitude());
                            jsonObject.put("direction", mobilePosition.getDirection());
                            jsonObject.put("speed", mobilePosition.getSpeed());
                            redisCatchStorage.sendMobilePositionMsg(jsonObject);
                        }
                    }
                    if (!ObjectUtils.isEmpty(deviceAlarm.getDeviceId())) {
                        if (deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.Video.getVal() + "")) {
                            deviceAlarm.setAlarmType(getText(sipMsgInfo.getRootElement().element("Info"), "AlarmType"));
                        if (!ObjectUtils.isEmpty(deviceAlarm.getDeviceId())) {
                            if (deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.Video.getVal() + "")) {
                                deviceAlarm.setAlarmType(getText(sipMsgInfo.getRootElement().element("Info"), "AlarmType"));
                            }
                        }
                    }
                    logger.info("[收到报警通知]内容：{}", JSON.toJSONString(deviceAlarm));
                    if ("7".equals(deviceAlarm.getAlarmMethod()) ) {
                        // 发送给平台的报警信息。 发送redis通知
                        AlarmChannelMessage alarmChannelMessage = new AlarmChannelMessage();
                        alarmChannelMessage.setAlarmSn(Integer.parseInt(deviceAlarm.getAlarmMethod()));
                        alarmChannelMessage.setAlarmDescription(deviceAlarm.getAlarmDescription());
                        alarmChannelMessage.setGbId(channelId);
                        redisCatchStorage.sendAlarmMsg(alarmChannelMessage);
                        continue;
                    }
                        logger.info("[收到报警通知]内容：{}", JSON.toJSONString(deviceAlarm));
                        if ("7".equals(deviceAlarm.getAlarmMethod()) ) {
                            // 发送给平台的报警信息。 发送redis通知
                            AlarmChannelMessage alarmChannelMessage = new AlarmChannelMessage();
                            alarmChannelMessage.setAlarmSn(Integer.parseInt(deviceAlarm.getAlarmMethod()));
                            alarmChannelMessage.setAlarmDescription(deviceAlarm.getAlarmDescription());
                            alarmChannelMessage.setGbId(channelId);
                            redisCatchStorage.sendAlarmMsg(alarmChannelMessage);
                            continue;
                        }

                    logger.debug("存储报警信息、报警分类");
                    // 存储报警信息、报警分类
                    if (sipConfig.isAlarm()) {
                        deviceAlarmService.add(deviceAlarm);
                    }
                        logger.debug("存储报警信息、报警分类");
                        // 存储报警信息、报警分类
                        if (sipConfig.isAlarm()) {
                            deviceAlarmService.add(deviceAlarm);
                        }

                    if (redisCatchStorage.deviceIsOnline(sipMsgInfo.getDevice().getDeviceId())) {
                        publisher.deviceAlarmEventPublish(deviceAlarm);
                        if (redisCatchStorage.deviceIsOnline(sipMsgInfo.getDevice().getDeviceId())) {
                            publisher.deviceAlarmEventPublish(deviceAlarm);
                        }
                    }catch (Exception e) {
                        logger.warn("[收到报警通知] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }


    }

    @Override

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java

@@ -22,7 +22,6 @@
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
@@ -57,8 +56,6 @@
    @Autowired
    private IDeviceChannelService deviceChannelService;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<SipMsgInfo> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
@@ -73,21 +70,22 @@
    @Override
    public void handForDevice(RequestEvent evt, Device device, Element rootElement) {

        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(new SipMsgInfo(evt, device, rootElement));
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        // 回复200 OK
        try {
            responseAck((SIPRequest) evt.getRequest(), Response.OK);
        } catch (SipException | InvalidArgumentException | ParseException e) {
            logger.error("[命令发送失败] 移动位置通知回复: {}", e.getMessage());
        }
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    SipMsgInfo sipMsgInfo = taskQueue.poll();
                    try {
                        Element rootElementAfterCharset = getRootElement(sipMsgInfo.getEvt(), sipMsgInfo.getDevice().getCharset());
                        if (rootElementAfterCharset == null) {
                            try {
                                logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest());
                                responseAck((SIPRequest) sipMsgInfo.getEvt().getRequest(), Response.BAD_REQUEST);
                            } catch (SipException | InvalidArgumentException | ParseException e) {
                                logger.error("[命令发送失败] 移动设备位置数据通知 内容为空: {}", e.getMessage());
                            }
                            logger.warn("[移动位置通知] {}处理失败，未识别到信息体", device.getDeviceId());
                            continue;
                        }
                        MobilePosition mobilePosition = new MobilePosition();
@@ -137,12 +135,6 @@
                            storager.insertMobilePosition(mobilePosition);
                        }
                        storager.updateChannelPosition(deviceChannel);
                        //回复 200 OK
                        try {
                            responseAck((SIPRequest) sipMsgInfo.getEvt().getRequest(), Response.OK);
                        } catch (SipException | InvalidArgumentException | ParseException e) {
                            logger.error("[命令发送失败] 移动设备位置数据回复200: {}", e.getMessage());
                        }

                        // 发送redis消息。 通知位置信息的变化
                        JSONObject jsonObject = new JSONObject();
@@ -158,14 +150,12 @@

                    } catch (DocumentException e) {
                        e.printStackTrace();
                    } catch (Exception e) {
                        logger.warn("[移动位置通知] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
                    }

                }
                taskQueueHandlerRun = false;
            });
        }


    }

    @Override

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java

@@ -1,18 +1,11 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd;

import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.CatalogDataCatch;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException;
@@ -24,7 +17,6 @@
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
@@ -45,12 +37,10 @@
    private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class);
    private final String cmdType = "Catalog";

    private boolean taskQueueHandlerRun = false;

    @Autowired
    private ResponseMessageHandler responseMessageHandler;

    private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
    private final ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();

    @Autowired
    private IVideoManagerStorage storager;
@@ -69,6 +59,7 @@

    @Override
    public void handForDevice(RequestEvent evt, Device device, Element element) {
        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(new HandlerCatchData(evt, device, element));
        // 回复200 OK
        try {
@@ -76,67 +67,71 @@
        } catch (SipException | InvalidArgumentException | ParseException e) {
            logger.error("[命令发送失败] 目录查询回复: {}", e.getMessage());
        }
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        // 如果不为空则说明已经开启消息处理
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    HandlerCatchData take = taskQueue.poll();
                    Element rootElement = null;
                    // 全局异常捕获，保证下一条可以得到处理
                    try {
                        rootElement = getRootElement(take.getEvt(), take.getDevice().getCharset());
                    } catch (DocumentException e) {
                        logger.error("[xml解析] 失败： ", e);
                        continue;
                    }
                    if (rootElement == null) {
                        logger.warn("[ 收到通道 ] content cannot be null, {}", evt.getRequest());
                        continue;
                    }
                    Element deviceListElement = rootElement.element("DeviceList");
                    Element sumNumElement = rootElement.element("SumNum");
                    Element snElement = rootElement.element("SN");
                    int sumNum = Integer.parseInt(sumNumElement.getText());

                    if (sumNum == 0) {
                        logger.info("[收到通道]设备:{}的: 0个", take.getDevice().getDeviceId());
                        // 数据已经完整接收
                        storager.cleanChannelsForDevice(take.getDevice().getDeviceId());
                        catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null);
                    } else {
                        Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
                        if (deviceListIterator != null) {
                            List<DeviceChannel> channelList = new ArrayList<>();
                            // 遍历DeviceList
                            while (deviceListIterator.hasNext()) {
                                Element itemDevice = deviceListIterator.next();
                                Element channelDeviceElement = itemDevice.element("DeviceID");
                                if (channelDeviceElement == null) {
                                    continue;
                                }
                                DeviceChannel deviceChannel = XmlUtil.channelContentHander(itemDevice, device, null);
                                deviceChannel.setDeviceId(take.getDevice().getDeviceId());

                                channelList.add(deviceChannel);
                            }
                            int sn = Integer.parseInt(snElement.getText());
                            catalogDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, take.getDevice(), channelList);
                            logger.info("[收到通道]设备: {} -> {}个，{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.get(take.getDevice().getDeviceId()) == null ? 0 : catalogDataCatch.get(take.getDevice().getDeviceId()).size(), sumNum);
                            if (catalogDataCatch.get(take.getDevice().getDeviceId()).size() == sumNum) {
                                // 数据已经完整接收， 此时可能存在某个设备离线变上线的情况，但是考虑到性能，此处不做处理，
                                // 目前支持设备通道上线通知时和设备上线时向上级通知
                                boolean resetChannelsResult = storager.resetChannels(take.getDevice().getDeviceId(), catalogDataCatch.get(take.getDevice().getDeviceId()));
                                if (!resetChannelsResult) {
                                    String errorMsg = "接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.get(take.getDevice().getDeviceId()).size() + "条";
                                    catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), errorMsg);
                                } else {
                                    catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null);
                                }
                            }
                        HandlerCatchData take = taskQueue.poll();
                        Element rootElement = null;
                        try {
                            rootElement = getRootElement(take.getEvt(), take.getDevice().getCharset());
                        } catch (DocumentException e) {
                            logger.error("[xml解析] 失败： ", e);
                            continue;
                        }
                        if (rootElement == null) {
                            logger.warn("[ 收到通道 ] content cannot be null, {}", evt.getRequest());
                            continue;
                        }
                        Element deviceListElement = rootElement.element("DeviceList");
                        Element sumNumElement = rootElement.element("SumNum");
                        Element snElement = rootElement.element("SN");
                        int sumNum = Integer.parseInt(sumNumElement.getText());

                        if (sumNum == 0) {
                            logger.info("[收到通道]设备:{}的: 0个", take.getDevice().getDeviceId());
                            // 数据已经完整接收
                            storager.cleanChannelsForDevice(take.getDevice().getDeviceId());
                            catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null);
                        } else {
                            Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
                            if (deviceListIterator != null) {
                                List<DeviceChannel> channelList = new ArrayList<>();
                                // 遍历DeviceList
                                while (deviceListIterator.hasNext()) {
                                    Element itemDevice = deviceListIterator.next();
                                    Element channelDeviceElement = itemDevice.element("DeviceID");
                                    if (channelDeviceElement == null) {
                                        continue;
                                    }
                                    DeviceChannel deviceChannel = XmlUtil.channelContentHander(itemDevice, device, null);
                                    deviceChannel.setDeviceId(take.getDevice().getDeviceId());

                                    channelList.add(deviceChannel);
                                }
                                int sn = Integer.parseInt(snElement.getText());
                                catalogDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, take.getDevice(), channelList);
                                logger.info("[收到通道]设备: {} -> {}个，{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.get(take.getDevice().getDeviceId()) == null ? 0 : catalogDataCatch.get(take.getDevice().getDeviceId()).size(), sumNum);
                                if (catalogDataCatch.get(take.getDevice().getDeviceId()).size() == sumNum) {
                                    // 数据已经完整接收， 此时可能存在某个设备离线变上线的情况，但是考虑到性能，此处不做处理，
                                    // 目前支持设备通道上线通知时和设备上线时向上级通知
                                    boolean resetChannelsResult = storager.resetChannels(take.getDevice().getDeviceId(), catalogDataCatch.get(take.getDevice().getDeviceId()));
                                    if (!resetChannelsResult) {
                                        String errorMsg = "接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.get(take.getDevice().getDeviceId()).size() + "条";
                                        catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), errorMsg);
                                    } else {
                                        catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null);
                                    }
                                }
                            }

                        }
                    }catch (Exception e) {
                        logger.warn("[收到通道] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java

@@ -9,7 +9,6 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException;
import org.dom4j.Element;
@@ -21,17 +20,17 @@
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;

import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText;

@@ -46,7 +45,6 @@

    private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();

    private boolean taskQueueHandlerRun = false;
    @Autowired
    private ResponseMessageHandler responseMessageHandler;

@@ -70,6 +68,7 @@

    @Override
    public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
        boolean isEmpty = taskQueue.isEmpty();
        try {
            // 回复200 OK
             responseAck((SIPRequest) evt.getRequest(), Response.OK);
@@ -77,8 +76,7 @@
            logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage());
        }
        taskQueue.offer(new HandlerCatchData(evt, device, rootElement));
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(()->{
                while (!taskQueue.isEmpty()) {
                    try {
@@ -151,9 +149,10 @@
                        }
                    } catch (DocumentException e) {
                        logger.error("xml解析异常： ", e);
                    } catch (Exception e) {
                        logger.warn("[国标录像] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }
    }

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisAlarmMsgListener.java

@@ -38,8 +38,6 @@
    @Autowired
    private IVideoManagerStorage storage;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
@@ -49,69 +47,68 @@
    @Override
    public void onMessage(@NotNull Message message, byte[] bytes) {
        logger.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody()));

        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(message);
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            logger.info("[线程池信息]活动线程数：{}, 最大线程数： {}", taskExecutor.getActiveCount(), taskExecutor.getMaxPoolSize());
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    try {
                        AlarmChannelMessage alarmChannelMessage = JSON.parseObject(msg.getBody(), AlarmChannelMessage.class);
                        if (alarmChannelMessage == null) {
                            logger.warn("[REDIS的ALARM通知]消息解析失败");
                            continue;
                        }
                        String gbId = alarmChannelMessage.getGbId();

                    AlarmChannelMessage alarmChannelMessage = JSON.parseObject(msg.getBody(), AlarmChannelMessage.class);
                    if (alarmChannelMessage == null) {
                        logger.warn("[REDIS的ALARM通知]消息解析失败");
                        continue;
                    }
                    String gbId = alarmChannelMessage.getGbId();
                        DeviceAlarm deviceAlarm = new DeviceAlarm();
                        deviceAlarm.setCreateTime(DateUtil.getNow());
                        deviceAlarm.setChannelId(gbId);
                        deviceAlarm.setAlarmDescription(alarmChannelMessage.getAlarmDescription());
                        deviceAlarm.setAlarmMethod("" + alarmChannelMessage.getAlarmSn());
                        deviceAlarm.setAlarmPriority("1");
                        deviceAlarm.setAlarmTime(DateUtil.getNowForISO8601());
                        deviceAlarm.setAlarmType("1");
                        deviceAlarm.setLongitude(0);
                        deviceAlarm.setLatitude(0);

                    DeviceAlarm deviceAlarm = new DeviceAlarm();
                    deviceAlarm.setCreateTime(DateUtil.getNow());
                    deviceAlarm.setChannelId(gbId);
                    deviceAlarm.setAlarmDescription(alarmChannelMessage.getAlarmDescription());
                    deviceAlarm.setAlarmMethod("" + alarmChannelMessage.getAlarmSn());
                    deviceAlarm.setAlarmPriority("1");
                    deviceAlarm.setAlarmTime(DateUtil.getNowForISO8601());
                    deviceAlarm.setAlarmType("1");
                    deviceAlarm.setLongitude(0);
                    deviceAlarm.setLatitude(0);

                    if (ObjectUtils.isEmpty(gbId)) {
                        // 发送给所有的上级
                        List<ParentPlatform> parentPlatforms = storage.queryEnableParentPlatformList(true);
                        if (parentPlatforms.size() > 0) {
                            for (ParentPlatform parentPlatform : parentPlatforms) {
                                try {
                                    commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm);
                                } catch (SipException | InvalidArgumentException | ParseException e) {
                                    logger.error("[命令发送失败] 国标级联 发送报警: {}", e.getMessage());
                        if (ObjectUtils.isEmpty(gbId)) {
                            // 发送给所有的上级
                            List<ParentPlatform> parentPlatforms = storage.queryEnableParentPlatformList(true);
                            if (parentPlatforms.size() > 0) {
                                for (ParentPlatform parentPlatform : parentPlatforms) {
                                    try {
                                        commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm);
                                    } catch (SipException | InvalidArgumentException | ParseException e) {
                                        logger.error("[命令发送失败] 国标级联 发送报警: {}", e.getMessage());
                                    }
                                }
                            }
                        }
                    }else {
                        Device device = storage.queryVideoDevice(gbId);
                        ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId);
                        if (device != null && platform == null) {
                            try {
                                commander.sendAlarmMessage(device, deviceAlarm);
                            } catch (InvalidArgumentException | SipException | ParseException e) {
                                logger.error("[命令发送失败] 发送报警: {}", e.getMessage());
                            }
                        }else if (device == null && platform != null){
                            try {
                                commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
                            } catch (InvalidArgumentException | SipException | ParseException e) {
                                logger.error("[命令发送失败] 发送报警: {}", e.getMessage());
                            }
                        }else {
                            logger.warn("无法确定" + gbId + "是平台还是设备");
                            Device device = storage.queryVideoDevice(gbId);
                            ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId);
                            if (device != null && platform == null) {
                                try {
                                    commander.sendAlarmMessage(device, deviceAlarm);
                                } catch (InvalidArgumentException | SipException | ParseException e) {
                                    logger.error("[命令发送失败] 发送报警: {}", e.getMessage());
                                }
                            }else if (device == null && platform != null){
                                try {
                                    commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
                                } catch (InvalidArgumentException | SipException | ParseException e) {
                                    logger.error("[命令发送失败] 发送报警: {}", e.getMessage());
                                }
                            }else {
                                logger.warn("无法确定" + gbId + "是平台还是设备");
                            }
                        }
                    }catch (Exception e) {
                        logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }


    }
}

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java

@@ -88,8 +88,6 @@
    @Autowired
    private ZlmHttpHookSubscribe subscribe;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
@@ -111,107 +109,104 @@

    @Override
    public void onMessage(Message message, byte[] bytes) {

        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(message);
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
                    WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
                    if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
                        continue;
                    }
                    if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
                        logger.info("[收到REDIS通知] 请求： {}", new String(msg.getBody()));

                        switch (wvpRedisMsg.getCmd()){
                            case WvpRedisMsgCmd.GET_SEND_ITEM:
                                RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
                                requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
                                break;
                            case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());;
                                requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
                                break;
                            default:
                                break;
                    try {
                        JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
                        WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
                        if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
                            continue;
                        }
                        if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
                            logger.info("[收到REDIS通知] 请求： {}", new String(msg.getBody()));

                    }else {
                        logger.info("[收到REDIS通知] 回复： {}", new String(msg.getBody()));
                        switch (wvpRedisMsg.getCmd()){
                            case WvpRedisMsgCmd.GET_SEND_ITEM:
                            switch (wvpRedisMsg.getCmd()){
                                case WvpRedisMsgCmd.GET_SEND_ITEM:
                                    RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
                                    requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
                                    break;
                                case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                    RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());
                                    requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
                                    break;
                                default:
                                    break;
                            }

                                WVPResult content  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
                        }else {
                            logger.info("[收到REDIS通知] 回复： {}", new String(msg.getBody()));
                            switch (wvpRedisMsg.getCmd()){
                                case WvpRedisMsgCmd.GET_SEND_ITEM:

                                String key = wvpRedisMsg.getSerial();
                                switch (content.getCode()) {
                                    case 0:
                                        ResponseSendItemMsg responseSendItemMsg =JSON.to(ResponseSendItemMsg.class, content.getData());
                                        PlayMsgCallback playMsgCallback = callbacks.get(key);
                                        if (playMsgCallback != null) {
                                            callbacksForError.remove(key);
                                            try {
                                                playMsgCallback.handler(responseSendItemMsg);
                                            } catch (ParseException e) {
                                                logger.error("[REDIS消息处理异常] ", e);
                                   WVPResult content  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());

                                    String key = wvpRedisMsg.getSerial();
                                    switch (content.getCode()) {
                                        case 0:
                                           ResponseSendItemMsg responseSendItemMsg =JSON.to(ResponseSendItemMsg.class, content.getData());
                                            PlayMsgCallback playMsgCallback = callbacks.get(key);
                                            if (playMsgCallback != null) {
                                                callbacksForError.remove(key);
                                                try {
                                                    playMsgCallback.handler(responseSendItemMsg);
                                                } catch (ParseException e) {
                                                    logger.error("[REDIS消息处理异常] ", e);
                                                }
                                            }
                                        }
                                        break;
                                    case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                    case ERROR_CODE_OFFLINE:
                                    case ERROR_CODE_TIMEOUT:
                                        PlayMsgErrorCallback errorCallback = callbacksForError.get(key);
                                        if (errorCallback != null) {
                                            callbacks.remove(key);
                                            errorCallback.handler(content);
                                        }
                                        break;
                                    default:
                                        break;
                                }
                                break;
                            case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                WVPResult wvpResult  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
                                String serial = wvpRedisMsg.getSerial();
                                switch (wvpResult.getCode()) {
                                    case 0:
                                        JSONObject jsonObject = (JSONObject)wvpResult.getData();
                                        PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial);
                                        if (playMsgCallback != null) {
                                            callbacksForError.remove(serial);
                                            playMsgCallback.handler(jsonObject);
                                        }
                                        break;
                                    case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                    case ERROR_CODE_OFFLINE:
                                    case ERROR_CODE_TIMEOUT:
                                        PlayMsgErrorCallback errorCallback = callbacksForError.get(serial);
                                        if (errorCallback != null) {
                                            callbacks.remove(serial);
                                            errorCallback.handler(wvpResult);
                                        }
                                        break;
                                    default:
                                        break;
                                }
                                break;
                            default:
                                break;
                                            break;
                                        case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                        case ERROR_CODE_OFFLINE:
                                        case ERROR_CODE_TIMEOUT:
                                            PlayMsgErrorCallback errorCallback = callbacksForError.get(key);
                                            if (errorCallback != null) {
                                                callbacks.remove(key);
                                                errorCallback.handler(content);
                                            }
                                            break;
                                        default:
                                            break;
                                    }
                                    break;
                                case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                    WVPResult wvpResult  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
                                    String serial = wvpRedisMsg.getSerial();
                                    switch (wvpResult.getCode()) {
                                        case 0:
                                            JSONObject jsonObject = (JSONObject)wvpResult.getData();
                                            PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial);
                                            if (playMsgCallback != null) {
                                                callbacksForError.remove(serial);
                                                playMsgCallback.handler(jsonObject);
                                            }
                                            break;
                                        case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                        case ERROR_CODE_OFFLINE:
                                        case ERROR_CODE_TIMEOUT:
                                            PlayMsgErrorCallback errorCallback = callbacksForError.get(serial);
                                            if (errorCallback != null) {
                                                callbacks.remove(serial);
                                                errorCallback.handler(wvpResult);
                                            }
                                            break;
                                        default:
                                            break;
                                    }
                                    break;
                                default:
                                    break;
                            }

                        }
                    }catch (Exception e) {
                        logger.warn("[RedisGbPlayMsg] 发现未处理的异常, {}",e.getMessage());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }






    }

    /**

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGpsMsgListener.java

@@ -27,8 +27,6 @@

    private final static Logger logger = LoggerFactory.getLogger(RedisGpsMsgListener.class);

    private boolean taskQueueHandlerRun = false;

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

@@ -44,17 +42,20 @@

    @Override
    public void onMessage(@NotNull Message message, byte[] bytes) {
        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(message);
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    GPSMsgInfo gpsMsgInfo = JSON.parseObject(msg.getBody(), GPSMsgInfo.class);
                    // 只是放入redis缓存起来
                    redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo);
                    try {
                        GPSMsgInfo gpsMsgInfo = JSON.parseObject(msg.getBody(), GPSMsgInfo.class);
                        // 只是放入redis缓存起来
                        redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo);
                    }catch (Exception e) {
                        logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }
    }

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java

@@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.service.redisMsg;

import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,8 +25,6 @@

    private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class);

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
@@ -43,24 +40,27 @@

    @Override
    public void onMessage(Message message, byte[] bytes) {
        logger.warn("[REDIS消息-请求推流结果]： {}", new String(message.getBody()));
        logger.info("[REDIS消息-请求推流结果]： {}", new String(message.getBody()));
        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(message);
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
                    if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
                        logger.info("[REDIS消息-请求推流结果]：参数不全");
                        continue;
                    }
                    // 查看正在等待的invite消息
                    if (responseEvents.get(response.getApp() + response.getStream()) != null) {
                        responseEvents.get(response.getApp() + response.getStream()).run(response);
                    try {
                        MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
                        if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
                            logger.info("[REDIS消息-请求推流结果]：参数不全");
                            continue;
                        }
                        // 查看正在等待的invite消息
                        if (responseEvents.get(response.getApp() + response.getStream()) != null) {
                            responseEvents.get(response.getApp() + response.getStream()).run(response);
                        }
                    }catch (Exception e) {
                        logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }
    }

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusListMsgListener.java

@@ -6,7 +6,6 @@
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.utils.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -18,7 +17,8 @@
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
@@ -38,7 +38,6 @@
    @Resource
    private IGbStreamService gbStreamService;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();

@@ -49,54 +48,56 @@
    @Override
    public void onMessage(Message message, byte[] bytes) {
        logger.info("[REDIS消息-推流设备列表更新]： {}", new String(message.getBody()));

        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(message);
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    List<StreamPushItem> streamPushItems = JSON.parseArray(new String(msg.getBody()), StreamPushItem.class);
                    //查询全部的app+stream 用于判断是添加还是修改
                    List<String> allAppAndStream = streamPushService.getAllAppAndStream();
                    try {
                        List<StreamPushItem> streamPushItems = JSON.parseArray(new String(msg.getBody()), StreamPushItem.class);
                        //查询全部的app+stream 用于判断是添加还是修改
                        List<String> allAppAndStream = streamPushService.getAllAppAndStream();

                    /**
                     * 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表
                     */
                    List<StreamPushItem> streamPushItemForSave = new ArrayList<>();
                    List<StreamPushItem> streamPushItemForUpdate = new ArrayList<>();
                    for (StreamPushItem streamPushItem : streamPushItems) {
                        String app = streamPushItem.getApp();
                        String stream = streamPushItem.getStream();
                        boolean contains = allAppAndStream.contains(app + stream);
                        //不存在就添加
                        if (!contains) {
                            streamPushItem.setStreamType("push");
                            streamPushItem.setCreateTime(DateUtil.getNow());
                            streamPushItem.setMediaServerId(mediaServerService.getDefaultMediaServer().getId());
                            streamPushItem.setOriginType(2);
                            streamPushItem.setOriginTypeStr("rtsp_push");
                            streamPushItem.setTotalReaderCount("0");
                            streamPushItemForSave.add(streamPushItem);
                        } else {
                            //存在就只修改 name和gbId
                            streamPushItemForUpdate.add(streamPushItem);
                        /**
                         * 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表
                         */
                        List<StreamPushItem> streamPushItemForSave = new ArrayList<>();
                        List<StreamPushItem> streamPushItemForUpdate = new ArrayList<>();
                        for (StreamPushItem streamPushItem : streamPushItems) {
                            String app = streamPushItem.getApp();
                            String stream = streamPushItem.getStream();
                            boolean contains = allAppAndStream.contains(app + stream);
                            //不存在就添加
                            if (!contains) {
                                streamPushItem.setStreamType("push");
                                streamPushItem.setCreateTime(DateUtil.getNow());
                                streamPushItem.setMediaServerId(mediaServerService.getDefaultMediaServer().getId());
                                streamPushItem.setOriginType(2);
                                streamPushItem.setOriginTypeStr("rtsp_push");
                                streamPushItem.setTotalReaderCount("0");
                                streamPushItemForSave.add(streamPushItem);
                            } else {
                                //存在就只修改 name和gbId
                                streamPushItemForUpdate.add(streamPushItem);
                            }
                        }
                    }
                    if (streamPushItemForSave.size() > 0) {
                        if (streamPushItemForSave.size() > 0) {

                        logger.info("添加{}条",streamPushItemForSave.size());
                        logger.info(JSONObject.toJSONString(streamPushItemForSave));
                        streamPushService.batchAdd(streamPushItemForSave);
                            logger.info("添加{}条",streamPushItemForSave.size());
                            logger.info(JSONObject.toJSONString(streamPushItemForSave));
                            streamPushService.batchAdd(streamPushItemForSave);

                    }
                    if(streamPushItemForUpdate.size()>0){
                        logger.info("修改{}条",streamPushItemForUpdate.size());
                        logger.info(JSONObject.toJSONString(streamPushItemForUpdate));
                        gbStreamService.updateGbIdOrName(streamPushItemForUpdate);
                        }
                        if(streamPushItemForUpdate.size()>0){
                            logger.info("修改{}条",streamPushItemForUpdate.size());
                            logger.info(JSONObject.toJSONString(streamPushItemForUpdate));
                            gbStreamService.updateGbIdOrName(streamPushItemForUpdate);
                        }
                    }catch (Exception e) {
                        logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }
    }

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java

@@ -29,8 +29,6 @@

    private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusMsgListener.class);

    private boolean taskQueueHandlerRun = false;

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

@@ -50,37 +48,40 @@

    @Override
    public void onMessage(Message message, byte[] bytes) {
        boolean isEmpty = taskQueue.isEmpty();
        logger.warn("[REDIS消息-推流设备状态变化]： {}", new String(message.getBody()));
        taskQueue.offer(message);

        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class);
                    if (statusChangeFromPushStream == null) {
                        logger.warn("[REDIS消息]推流设备状态变化消息解析失败");
                        continue;
                    }
                    // 取消定时任务
                    dynamicTask.stop(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED);
                    if (statusChangeFromPushStream.isSetAllOffline()) {
                        // 所有设备离线
                        streamPushService.allStreamOffline();
                    }
                    if (statusChangeFromPushStream.getOfflineStreams() != null
                            && statusChangeFromPushStream.getOfflineStreams().size() > 0) {
                        // 更新部分设备离线
                        streamPushService.offline(statusChangeFromPushStream.getOfflineStreams());
                    }
                    if (statusChangeFromPushStream.getOnlineStreams() != null &&
                            statusChangeFromPushStream.getOnlineStreams().size() > 0) {
                        // 更新部分设备上线
                        streamPushService.online(statusChangeFromPushStream.getOnlineStreams());
                    try {
                        PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class);
                        if (statusChangeFromPushStream == null) {
                            logger.warn("[REDIS消息]推流设备状态变化消息解析失败");
                            continue;
                        }
                        // 取消定时任务
                        dynamicTask.stop(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED);
                        if (statusChangeFromPushStream.isSetAllOffline()) {
                            // 所有设备离线
                            streamPushService.allStreamOffline();
                        }
                        if (statusChangeFromPushStream.getOfflineStreams() != null
                                && statusChangeFromPushStream.getOfflineStreams().size() > 0) {
                            // 更新部分设备离线
                            streamPushService.offline(statusChangeFromPushStream.getOfflineStreams());
                        }
                        if (statusChangeFromPushStream.getOnlineStreams() != null &&
                                statusChangeFromPushStream.getOnlineStreams().size() > 0) {
                            // 更新部分设备上线
                            streamPushService.online(statusChangeFromPushStream.getOnlineStreams());
                        }
                    }catch (Exception e) {
                        logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }
    }

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java

@@ -33,8 +33,6 @@
    @Autowired
    private ZLMMediaListManager zlmMediaListManager;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
@@ -43,48 +41,50 @@

    @Override
    public void onMessage(Message message, byte[] bytes) {

        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(message);
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
                    if (steamMsgJson == null) {
                        logger.warn("[收到redis 流变化]消息解析失败");
                        continue;
                    }
                    String serverId = steamMsgJson.getString("serverId");
                    try {
                        JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
                        if (steamMsgJson == null) {
                            logger.warn("[收到redis 流变化]消息解析失败");
                            continue;
                        }
                        String serverId = steamMsgJson.getString("serverId");

                    if (userSetting.getServerId().equals(serverId)) {
                        // 自己发送的消息忽略即可
                        continue;
                    }
                    logger.info("[收到redis 流变化]： {}", new String(message.getBody()));
                    String app = steamMsgJson.getString("app");
                    String stream = steamMsgJson.getString("stream");
                    boolean register = steamMsgJson.getBoolean("register");
                    String mediaServerId = steamMsgJson.getString("mediaServerId");
                    OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
                    onStreamChangedHookParam.setSeverId(serverId);
                    onStreamChangedHookParam.setApp(app);
                    onStreamChangedHookParam.setStream(stream);
                    onStreamChangedHookParam.setRegist(register);
                    onStreamChangedHookParam.setMediaServerId(mediaServerId);
                    onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
                    onStreamChangedHookParam.setAliveSecond(0L);
                    onStreamChangedHookParam.setTotalReaderCount("0");
                    onStreamChangedHookParam.setOriginType(0);
                    onStreamChangedHookParam.setOriginTypeStr("0");
                    onStreamChangedHookParam.setOriginTypeStr("unknown");
                    if (register) {
                        zlmMediaListManager.addPush(onStreamChangedHookParam);
                    }else {
                        zlmMediaListManager.removeMedia(app, stream);
                        if (userSetting.getServerId().equals(serverId)) {
                            // 自己发送的消息忽略即可
                            continue;
                        }
                        logger.info("[收到redis 流变化]： {}", new String(message.getBody()));
                        String app = steamMsgJson.getString("app");
                        String stream = steamMsgJson.getString("stream");
                        boolean register = steamMsgJson.getBoolean("register");
                        String mediaServerId = steamMsgJson.getString("mediaServerId");
                        OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
                        onStreamChangedHookParam.setSeverId(serverId);
                        onStreamChangedHookParam.setApp(app);
                        onStreamChangedHookParam.setStream(stream);
                        onStreamChangedHookParam.setRegist(register);
                        onStreamChangedHookParam.setMediaServerId(mediaServerId);
                        onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
                        onStreamChangedHookParam.setAliveSecond(0L);
                        onStreamChangedHookParam.setTotalReaderCount("0");
                        onStreamChangedHookParam.setOriginType(0);
                        onStreamChangedHookParam.setOriginTypeStr("0");
                        onStreamChangedHookParam.setOriginTypeStr("unknown");
                        if (register) {
                            zlmMediaListManager.addPush(onStreamChangedHookParam);
                        }else {
                            zlmMediaListManager.removeMedia(app, stream);
                        }
                    }catch (Exception e) {
                        logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
                    }
                }
                taskQueueHandlerRun = false;
            });
        }
    }

 src/main/resources/all-application.yml

@@ -150,8 +150,6 @@
        enable: true
        # [可选] 在此范围内选择端口用于媒体流传输, 必须提前在zlm上配置该属性，不然自动配置此属性可能不成功
        port-range: 30000,30500 # 端口范围
        # [可选] 国标级联在此范围内选择端口发送媒体流
        send-port-range: 30000,30500 # 端口范围
    # 录像辅助服务， 部署此服务可以实现zlm录像的管理与下载， 0 表示不使用
    record-assist-port: 0