添加队列处理redis消息和sip消息，支持使用推流状态作为通道在线状态

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

@@ -0,0 +1,100 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson.JSON;
+import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.utils.DateUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ObjectUtils;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+
+@Component
+public class RedisAlarmMsgListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisAlarmMsgListener.class);
+
+    @Autowired
+    private ISIPCommander commander;
+
+    @Autowired
+    private ISIPCommanderForPlatform commanderForPlatform;
+
+    @Autowired
+    private IVideoManagerStorage storage;
+
+    private boolean taskQueueHandlerRun = false;
+
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody()));
+
+        taskQueue.offer(message);
+        if (!taskQueueHandlerRun) {
+            taskQueueHandlerRun = true;
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+
+                    AlarmChannelMessage alarmChannelMessage = JSON.parseObject(msg.getBody(), AlarmChannelMessage.class);
+                    if (alarmChannelMessage == null) {
+                        logger.warn("[REDIS的ALARM通知]消息解析失败");
+                        return;
+                    }
+                    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);
+
+                    if (ObjectUtils.isEmpty(gbId)) {
+                        // 发送给所有的上级
+                        List<ParentPlatform> parentPlatforms = storage.queryEnableParentPlatformList(true);
+                        if (parentPlatforms.size() > 0) {
+                            for (ParentPlatform parentPlatform : parentPlatforms) {
+                                commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm);
+                            }
+                        }
+                    }else {
+                        Device device = storage.queryVideoDevice(gbId);
+                        ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId);
+                        if (device != null && platform == null) {
+                            commander.sendAlarmMessage(device, deviceAlarm);
+                        }else if (device == null && platform != null){
+                            commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
+                        }else {
+                            logger.warn("无法确定" + gbId + "是平台还是设备");
+                        }
+                    }
+                }
+                taskQueueHandlerRun = false;
+            });
+        }
+
+
+    }
+}

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

@@ -0,0 +1,398 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.conf.DynamicTask;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
+import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
+import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.service.IMediaServerService;
+import com.genersoft.iot.vmp.service.bean.*;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.utils.redis.RedisUtil;
+import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.text.ParseException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+
+/**
+ * 监听下级发送推送信息，并发送国标推流消息上级
+ * @author lin
+ */
+@Component
+public class RedisGbPlayMsgListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisGbPlayMsgListener.class);
+
+    public static final String WVP_PUSH_STREAM_KEY = "WVP_PUSH_STREAM";
+
+    /**
+     * 流媒体不存在的错误玛
+     */
+    public static final  int ERROR_CODE_MEDIA_SERVER_NOT_FOUND = -1;
+
+    /**
+     * 离线的错误玛
+     */
+    public static final  int ERROR_CODE_OFFLINE = -2;
+
+    /**
+     * 超时的错误玛
+     */
+    public static final  int ERROR_CODE_TIMEOUT = -3;
+
+    private Map<String, PlayMsgCallback> callbacks = new ConcurrentHashMap<>();
+    private Map<String, PlayMsgCallbackForStartSendRtpStream> callbacksForStartSendRtpStream = new ConcurrentHashMap<>();
+    private Map<String, PlayMsgErrorCallback> callbacksForError = new ConcurrentHashMap<>();
+
+    @Autowired
+    private UserSetting userSetting;
+
+
+    @Autowired
+    private ZLMMediaListManager zlmMediaListManager;
+
+    @Autowired
+    private ZLMRTPServerFactory zlmrtpServerFactory;
+
+    @Autowired
+    private IMediaServerService mediaServerService;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private DynamicTask dynamicTask;
+
+    @Autowired
+    private ZLMMediaListManager mediaListManager;
+
+    @Autowired
+    private ZlmHttpHookSubscribe subscribe;
+
+    private boolean taskQueueHandlerRun = false;
+
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+
+    public interface PlayMsgCallback{
+        void handler(ResponseSendItemMsg responseSendItemMsg) throws ParseException;
+    }
+
+    public interface PlayMsgCallbackForStartSendRtpStream{
+        void handler(JSONObject jsonObject);
+    }
+
+    public interface PlayMsgErrorCallback{
+        void handler(WVPResult wvpResult);
+    }
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+
+        taskQueue.offer(message);
+        if (!taskQueueHandlerRun) {
+            taskQueueHandlerRun = true;
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+                    JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
+                    WvpRedisMsg wvpRedisMsg = JSON.toJavaObject(msgJSON, WvpRedisMsg.class);
+                    if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
+                        return;
+                    }
+                    if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
+                        logger.info("[收到REDIS通知] 请求： {}", new String(msg.getBody()));
+
+                        switch (wvpRedisMsg.getCmd()){
+                            case WvpRedisMsgCmd.GET_SEND_ITEM:
+                                RequestSendItemMsg content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestSendItemMsg.class);
+                                requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
+                                break;
+                            case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
+                                RequestPushStreamMsg param = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestPushStreamMsg.class);;
+                                requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
+                                break;
+                            default:
+                                break;
+                        }
+
+                    }else {
+                        logger.info("[收到REDIS通知] 回复： {}", new String(msg.getBody()));
+                        switch (wvpRedisMsg.getCmd()){
+                            case WvpRedisMsgCmd.GET_SEND_ITEM:
+
+                                WVPResult content  = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class);
+
+                                String key = wvpRedisMsg.getSerial();
+                                switch (content.getCode()) {
+                                    case 0:
+                                        ResponseSendItemMsg responseSendItemMsg =JSON.toJavaObject((JSONObject)content.getData(), ResponseSendItemMsg.class);
+                                        PlayMsgCallback playMsgCallback = callbacks.get(key);
+                                        if (playMsgCallback != null) {
+                                            callbacksForError.remove(key);
+                                            try {
+                                                playMsgCallback.handler(responseSendItemMsg);
+                                            } catch (ParseException e) {
+                                                throw new RuntimeException(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.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class);
+                                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;
+                        }
+                    }
+                }
+                taskQueueHandlerRun = false;
+            });
+        }
+
+
+
+
+
+
+    }
+
+    /**
+     * 处理收到的请求推流的请求
+     */
+    private void requestPushStreamMsgHand(RequestPushStreamMsg requestPushStreamMsg, String fromId, String serial) {
+        MediaServerItem mediaInfo = mediaServerService.getOne(requestPushStreamMsg.getMediaServerId());
+        if (mediaInfo == null) {
+            // TODO 回复错误
+            return;
+        }
+        String is_Udp = requestPushStreamMsg.isTcp() ? "0" : "1";
+        Map<String, Object> param = new HashMap<>();
+        param.put("vhost","__defaultVhost__");
+        param.put("app",requestPushStreamMsg.getApp());
+        param.put("stream",requestPushStreamMsg.getStream());
+        param.put("ssrc", requestPushStreamMsg.getSsrc());
+        param.put("dst_url",requestPushStreamMsg.getIp());
+        param.put("dst_port", requestPushStreamMsg.getPort());
+        param.put("is_udp", is_Udp);
+        param.put("src_port", requestPushStreamMsg.getSrcPort());
+        param.put("pt", requestPushStreamMsg.getPt());
+        param.put("use_ps", requestPushStreamMsg.isPs() ? "1" : "0");
+        param.put("only_audio", requestPushStreamMsg.isOnlyAudio() ? "1" : "0");
+        JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+        // 回复消息
+        responsePushStream(jsonObject, fromId, serial);
+    }
+
+    private void responsePushStream(JSONObject content, String toId, String serial) {
+
+        WVPResult<JSONObject> result = new WVPResult<>();
+        result.setCode(0);
+        result.setData(content);
+
+        WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
+                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, result);
+        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
+        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+    }
+
+    /**
+     * 处理收到的请求sendItem的请求
+     */
+    private void requestSendItemMsgHand(RequestSendItemMsg content, String toId, String serial) {
+        MediaServerItem mediaServerItem = mediaServerService.getOne(content.getMediaServerId());
+        if (mediaServerItem == null) {
+            logger.info("[回复推流信息] 流媒体{}不存在 ", content.getMediaServerId());
+
+            WVPResult<SendRtpItem> result = new WVPResult<>();
+            result.setCode(ERROR_CODE_MEDIA_SERVER_NOT_FOUND);
+            result.setMsg("流媒体不存在");
+
+            WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
+                    WvpRedisMsgCmd.GET_SEND_ITEM, serial, result);
+
+            JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
+            RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+            return;
+        }
+        // 确定流是否在线
+        boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream());
+        if (streamReady) {
+            logger.info("[回复推流信息]  {}/{}", content.getApp(), content.getStream());
+            responseSendItem(mediaServerItem, content, toId, serial);
+        }else {
+            // 流已经离线
+            // 发送redis消息以使设备上线
+            logger.info("[ app={}, stream={} ]通道离线，发送redis信息控制设备开始推流",content.getApp(), content.getStream());
+
+            String taskKey = UUID.randomUUID().toString();
+            // 设置超时
+            dynamicTask.startDelay(taskKey, ()->{
+                logger.info("[ app={}, stream={} ] 等待设备开始推流超时", content.getApp(), content.getStream());
+                WVPResult<SendRtpItem> result = new WVPResult<>();
+                result.setCode(ERROR_CODE_TIMEOUT);
+                WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
+                        userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
+                );
+                JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
+                RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+            }, userSetting.getPlatformPlayTimeout());
+
+            // 添加订阅
+            HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtsp", mediaServerItem.getId());
+
+            subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
+                        dynamicTask.stop(taskKey);
+                        responseSendItem(mediaServerItem, content, toId, serial);
+                    });
+
+            MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1, content.getApp(), content.getStream(),
+                    content.getChannelId(), content.getPlatformId(), content.getPlatformName(), content.getServerId(),
+                    content.getMediaServerId());
+            redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
+
+        }
+    }
+
+    /**
+     * 将获取到的sendItem发送出去
+     */
+    private void responseSendItem(MediaServerItem mediaServerItem, RequestSendItemMsg content, String toId, String serial) {
+        SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
+                content.getPort(), content.getSsrc(), content.getPlatformId(),
+                content.getApp(), content.getStream(), content.getChannelId(),
+                content.getTcp());
+
+        WVPResult<ResponseSendItemMsg> result = new WVPResult<>();
+        result.setCode(0);
+        ResponseSendItemMsg responseSendItemMsg = new ResponseSendItemMsg();
+        responseSendItemMsg.setSendRtpItem(sendRtpItem);
+        responseSendItemMsg.setMediaServerItem(mediaServerItem);
+        result.setData(responseSendItemMsg);
+
+        WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
+                userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
+        );
+        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
+        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+    }
+
+    /**
+     * 发送消息要求下级生成推流信息
+     * @param serverId 下级服务ID
+     * @param app 应用名
+     * @param stream 流ID
+     * @param ip 目标IP
+     * @param port 目标端口
+     * @param ssrc  ssrc
+     * @param platformId 平台国标编号
+     * @param channelId 通道ID
+     * @param isTcp 是否使用TCP
+     * @param callback 得到信息的回调
+     */
+    public void sendMsg(String serverId, String mediaServerId, String app, String stream, String ip, int port, String ssrc,
+                        String platformId, String channelId, boolean isTcp, String platformName, PlayMsgCallback callback, PlayMsgErrorCallback errorCallback) {
+        RequestSendItemMsg requestSendItemMsg = RequestSendItemMsg.getInstance(
+                serverId, mediaServerId, app, stream, ip, port, ssrc, platformId, channelId, isTcp, platformName);
+        requestSendItemMsg.setServerId(serverId);
+        String key = UUID.randomUUID().toString();
+        WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId, WvpRedisMsgCmd.GET_SEND_ITEM,
+                key, requestSendItemMsg);
+
+        JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
+        logger.info("[请求推流SendItem] {}: {}", serverId, jsonObject);
+        callbacks.put(key, callback);
+        callbacksForError.put(key, errorCallback);
+        dynamicTask.startDelay(key, ()->{
+            callbacks.remove(key);
+            callbacksForError.remove(key);
+            WVPResult<Object> wvpResult = new WVPResult<>();
+            wvpResult.setCode(ERROR_CODE_TIMEOUT);
+            wvpResult.setMsg("timeout");
+            errorCallback.handler(wvpResult);
+        }, userSetting.getPlatformPlayTimeout());
+        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+    }
+
+    /**
+     * 发送请求推流的消息
+     * @param param 推流参数
+     * @param callback 回调
+     */
+    public void sendMsgForStartSendRtpStream(String serverId, RequestPushStreamMsg param, PlayMsgCallbackForStartSendRtpStream callback) {
+        String key = UUID.randomUUID().toString();
+        WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId,
+                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, param);
+
+        JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
+        logger.info("[REDIS 请求其他平台推流] {}: {}", serverId, jsonObject);
+        dynamicTask.startDelay(key, ()->{
+            callbacksForStartSendRtpStream.remove(key);
+            callbacksForError.remove(key);
+        }, userSetting.getPlatformPlayTimeout());
+        callbacksForStartSendRtpStream.put(key, callback);
+        callbacksForError.put(key, (wvpResult)->{
+            logger.info("[REDIS 请求其他平台推流] 失败: {}", wvpResult.getMsg());
+            callbacksForStartSendRtpStream.remove(key);
+            callbacksForError.remove(key);
+        });
+        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+    }
+}

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

@@ -0,0 +1,76 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson.JSON;
+import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * 接收来自redis的GPS更新通知
+ * @author lin
+ */
+@Component
+public class RedisGpsMsgListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisGpsMsgListener.class);
+
+    private boolean taskQueueHandlerRun = false;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private IVideoManagerStorage storager;
+
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+
+    @Override
+    public void onMessage(@NotNull Message message, byte[] bytes) {
+        taskQueue.offer(message);
+        if (!taskQueueHandlerRun) {
+            taskQueueHandlerRun = true;
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+                    GPSMsgInfo gpsMsgInfo = JSON.parseObject(msg.getBody(), GPSMsgInfo.class);
+                    // 只是放入redis缓存起来
+                    redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo);
+                }
+                taskQueueHandlerRun = false;
+            });
+        }
+    }
+
+    /**
+     * 定时将经纬度更新到数据库
+     */
+    @Scheduled(fixedRate = 2 * 1000)   //每2秒执行一次
+    public void execute(){
+        List<GPSMsgInfo> gpsMsgInfo = redisCatchStorage.getAllGpsMsgInfo();
+        if (gpsMsgInfo.size() > 0) {
+            storager.updateStreamGPS(gpsMsgInfo);
+            for (GPSMsgInfo msgInfo : gpsMsgInfo) {
+                msgInfo.setStored(true);
+                redisCatchStorage.updateGpsMsgInfo(msgInfo);
+            }
+        }
+    }
+}

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

@@ -0,0 +1,75 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson.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;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ObjectUtils;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * 接收redis返回的推流结果
+ * @author lin
+ */
+@Component
+public class RedisPushStreamResponseListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class);
+
+    private boolean taskQueueHandlerRun = false;
+
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+
+    private Map<String, PushStreamResponseEvent> responseEvents = new ConcurrentHashMap<>();
+
+    public interface PushStreamResponseEvent{
+        void run(MessageForPushChannelResponse response);
+    }
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.warn("[REDIS消息-请求推流结果]： {}", new String(message.getBody()));
+        taskQueue.offer(message);
+        if (!taskQueueHandlerRun) {
+            taskQueueHandlerRun = true;
+            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消息-请求推流结果]：参数不全");
+                        return;
+                    }
+                    // 查看正在等待的invite消息
+                    if (responseEvents.get(response.getApp() + response.getStream()) != null) {
+                        responseEvents.get(response.getApp() + response.getStream()).run(response);
+                    }
+                }
+                taskQueueHandlerRun = false;
+            });
+        }
+    }
+
+    public void addEvent(String app, String stream, PushStreamResponseEvent callback) {
+        responseEvents.put(app + stream, callback);
+    }
+
+    public void removeEvent(String app, String stream) {
+        responseEvents.remove(app + stream);
+    }
+}

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

@@ -0,0 +1,103 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+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;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.util.*;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * @Auther: JiangFeng
+ * @Date: 2022/8/16 11:32
+ * @Description: 接收redis发送的推流设备列表更新通知
+ */
+@Component
+public class RedisPushStreamStatusListMsgListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusListMsgListener.class);
+    @Resource
+    private IMediaServerService mediaServerService;
+
+    @Resource
+    private IStreamPushService streamPushService;
+    @Resource
+    private IGbStreamService gbStreamService;
+
+    private boolean taskQueueHandlerRun = false;
+
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.info("[REDIS消息-推流设备列表更新]： {}", new String(message.getBody()));
+
+        taskQueue.offer(message);
+        if (!taskQueueHandlerRun) {
+            taskQueueHandlerRun = true;
+            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();
+
+                    /**
+                     * 用于存储更具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) {
+
+                        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);
+                    }
+                }
+                taskQueueHandlerRun = false;
+            });
+        }
+    }
+}

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

@@ -0,0 +1,99 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson.JSON;
+import com.genersoft.iot.vmp.common.VideoManagerConstants;
+import com.genersoft.iot.vmp.conf.DynamicTask;
+import com.genersoft.iot.vmp.service.IStreamPushService;
+import com.genersoft.iot.vmp.service.bean.PushStreamStatusChangeFromRedisDto;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+
+/**
+ * 接收redis发送的推流设备上线下线通知
+ * @author lin
+ */
+@Component
+public class RedisPushStreamStatusMsgListener implements MessageListener, ApplicationRunner {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusMsgListener.class);
+
+    private boolean taskQueueHandlerRun = false;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private IStreamPushService streamPushService;
+
+    @Autowired
+    private DynamicTask dynamicTask;
+
+
+
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.warn("[REDIS消息-推流设备状态变化]： {}", new String(message.getBody()));
+        taskQueue.offer(message);
+
+        if (!taskQueueHandlerRun) {
+            taskQueueHandlerRun = true;
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+                    PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class);
+                    if (statusChangeFromPushStream == null) {
+                        logger.warn("[REDIS消息]推流设备状态变化消息解析失败");
+                        return;
+                    }
+                    // 取消定时任务
+                    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());
+                    }
+                }
+                taskQueueHandlerRun = false;
+            });
+        }
+    }
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        //  启动时设置所有推流通道离线，发起查询请求
+        redisCatchStorage.sendStreamPushRequestedMsgForStatus();
+        dynamicTask.startDelay(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED, ()->{
+            logger.info("[REDIS消息]未收到redis回复推流设备状态，执行推流设备离线");
+            // 五秒收不到请求就设置通道离线，然后通知上级离线
+            streamPushService.allStreamOffline();
+        }, 5000);
+    }
+
+}

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

@@ -0,0 +1,91 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.conf.UserSetting;
+
+import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+
+/**
+ * 接收其他wvp发送流变化通知
+ * @author lin
+ */
+@Component
+public class RedisStreamMsgListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisStreamMsgListener.class);
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @Autowired
+    private ZLMMediaListManager zlmMediaListManager;
+
+    private boolean taskQueueHandlerRun = false;
+
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+
+        taskQueue.offer(message);
+        if (!taskQueueHandlerRun) {
+            taskQueueHandlerRun = true;
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+                    JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
+                    if (steamMsgJson == null) {
+                        logger.warn("[收到redis 流变化]消息解析失败");
+                        return;
+                    }
+                    String serverId = steamMsgJson.getString("serverId");
+
+                    if (userSetting.getServerId().equals(serverId)) {
+                        // 自己发送的消息忽略即可
+                        return;
+                    }
+                    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");
+                    MediaItem mediaItem = new MediaItem();
+                    mediaItem.setSeverId(serverId);
+                    mediaItem.setApp(app);
+                    mediaItem.setStream(stream);
+                    mediaItem.setRegist(register);
+                    mediaItem.setMediaServerId(mediaServerId);
+                    mediaItem.setCreateStamp(System.currentTimeMillis()/1000);
+                    mediaItem.setAliveSecond(0L);
+                    mediaItem.setTotalReaderCount("0");
+                    mediaItem.setOriginType(0);
+                    mediaItem.setOriginTypeStr("0");
+                    mediaItem.setOriginTypeStr("unknown");
+                    if (register) {
+                        zlmMediaListManager.addPush(mediaItem);
+                    }else {
+                        zlmMediaListManager.removeMedia(app, stream);
+                    }
+                }
+                taskQueueHandlerRun = false;
+            });
+        }
+    }
+}