优化REDIS消息高并发处理 #1578

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

@@ -78,7 +78,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
 
         RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, userSetting.getSipUseSourceIpAsRemoteAddress());
         if (!device.getIp().equalsIgnoreCase(remoteAddressInfo.getIp()) || device.getPort() != remoteAddressInfo.getPort()) {
-            log.info("[收到心跳] 设备{}地址变化, 远程地址为: {}:{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort());
+            log.info("[收到心跳] 设备{}地址变化, {}:{}->{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort(), request.getLocalAddress().getHostAddress());
             device.setPort(remoteAddressInfo.getPort());
             device.setHostAddress(remoteAddressInfo.getIp().concat(":").concat(String.valueOf(remoteAddressInfo.getPort())));
             device.setIp(remoteAddressInfo.getIp());

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

@@ -15,10 +15,9 @@ import com.genersoft.iot.vmp.service.IMobilePositionService;
 import com.genersoft.iot.vmp.utils.DateUtil;
 import lombok.extern.slf4j.Slf4j;
 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.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ObjectUtils;
 
@@ -26,6 +25,7 @@ import javax.sip.InvalidArgumentException;
 import javax.sip.SipException;
 import javax.validation.constraints.NotNull;
 import java.text.ParseException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
@@ -55,25 +55,34 @@ public class RedisAlarmMsgListener implements MessageListener {
     @Autowired
     private IPlatformService platformService;
 
-    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
-
-    @Qualifier("taskExecutor")
-    @Autowired
-    private ThreadPoolTaskExecutor taskExecutor;
 
     @Autowired
     private UserSetting userSetting;
 
     @Override
     public void onMessage(@NotNull Message message, byte[] bytes) {
-        log.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody()));
+        log.info("[REDIS: ALARM]： {}", new String(message.getBody()));
-        boolean isEmpty = taskQueue.isEmpty();
         taskQueue.offer(message);
-        if (isEmpty) {
+    }
-//            logger.info("[线程池信息]活动线程数：{}, 最大线程数： {}", taskExecutor.getActiveCount(), taskExecutor.getMaxPoolSize());
+
-            taskExecutor.execute(() -> {
+    @Scheduled(fixedDelay = 100)
-                while (!taskQueue.isEmpty()) {
+    public void executeTaskQueue() {
+        if (taskQueue.isEmpty()) {
+            return;
+        }
+        List<Message> messageDataList = new ArrayList<>();
+        int size = taskQueue.size();
+        for (int i = 0; i < size; i++) {
             Message msg = taskQueue.poll();
+            if (msg != null) {
+                messageDataList.add(msg);
+            }
+        }
+        if (messageDataList.isEmpty()) {
+            return;
+        }
+        for (Message msg : messageDataList) {
             try {
                 AlarmChannelMessage alarmChannelMessage = JSON.parseObject(msg.getBody(), AlarmChannelMessage.class);
                 if (alarmChannelMessage == null) {
@@ -107,10 +116,10 @@ public class RedisAlarmMsgListener implements MessageListener {
                                 }
                             }
                         }
-                            }else {
+                    } else {
                         // 获取开启了消息推送的设备和平台
                         List<Platform> parentPlatforms = mobilePositionService.queryEnablePlatformListWithAsMessageChannel();
-                                if (parentPlatforms.size() > 0) {
+                        if (!parentPlatforms.isEmpty()) {
                             for (Platform parentPlatform : parentPlatforms) {
                                 try {
                                     deviceAlarm.setChannelId(parentPlatform.getDeviceGBId());
@@ -124,7 +133,7 @@ public class RedisAlarmMsgListener implements MessageListener {
                     }
                     // 获取开启了消息推送的设备和平台
                     List<Device> devices = channelService.queryDeviceWithAsMessageChannel();
-                            if (devices.size() > 0) {
+                    if (!devices.isEmpty()) {
                         for (Device device : devices) {
                             try {
                                 deviceAlarm.setChannelId(device.getDeviceId());
@@ -135,7 +144,7 @@ public class RedisAlarmMsgListener implements MessageListener {
                         }
                     }
 
-                        }else {
+                } else {
                     Device device = deviceService.getDeviceByDeviceId(gbId);
                     Platform platform = platformService.queryPlatformByServerGBId(gbId);
                     if (device != null && platform == null) {
@@ -144,22 +153,21 @@ public class RedisAlarmMsgListener implements MessageListener {
                         } catch (InvalidArgumentException | SipException | ParseException e) {
                             log.error("[命令发送失败] 发送报警: {}", e.getMessage());
                         }
-                            }else if (device == null && platform != null){
+                    } else if (device == null && platform != null) {
                         try {
                             commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
                         } catch (InvalidArgumentException | SipException | ParseException e) {
                             log.error("[命令发送失败] 发送报警: {}", e.getMessage());
                         }
-                            }else {
+                    } else {
                         log.warn("无法确定" + gbId + "是平台还是设备");
                     }
                 }
-                    }catch (Exception e) {
+            } catch (Exception e) {
                 log.error("未处理的异常 ", e);
-                        log.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
+                log.warn("[REDIS的ALARM通知] 发现未处理的异常, {}", e.getMessage());
             }
         }
-            });
-        }
     }
 }
+

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

@@ -6,12 +6,13 @@ import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 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.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 /**
@@ -26,31 +27,40 @@ public class RedisCloseStreamMsgListener implements MessageListener {
     @Autowired
     private IStreamPushService pushService;
 
-    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
-
-    @Qualifier("taskExecutor")
-    @Autowired
-    private ThreadPoolTaskExecutor taskExecutor;
 
     @Override
     public void onMessage(@NotNull Message message, byte[] bytes) {
-        boolean isEmpty = taskQueue.isEmpty();
+        log.info("[REDIS: 关闭流]： {}", new String(message.getBody()));
         taskQueue.offer(message);
-        if (isEmpty) {
+    }
-            taskExecutor.execute(() -> {
+
-                while (!taskQueue.isEmpty()) {
+    @Scheduled(fixedDelay = 100)
+    public void executeTaskQueue() {
+        if (taskQueue.isEmpty()) {
+            return;
+        }
+        List<Message> messageDataList = new ArrayList<>();
+        int size = taskQueue.size();
+        for (int i = 0; i < size; i++) {
             Message msg = taskQueue.poll();
+            if (msg != null) {
+                messageDataList.add(msg);
+            }
+        }
+        if (messageDataList.isEmpty()) {
+            return;
+        }
+        for (Message msg : messageDataList) {
             try {
                 JSONObject jsonObject = JSON.parseObject(msg.getBody());
                 String app = jsonObject.getString("app");
                 String stream = jsonObject.getString("stream");
                 pushService.stopByAppAndStream(app, stream);
             }catch (Exception e) {
-                        log.warn("[REDIS的关闭推流通知] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
+                log.warn("[REDIS的关闭推流通知] 发现未处理的异常, \r\n{}", JSON.toJSONString(msg));
                 log.error("[REDIS的关闭推流通知] 异常内容： ", e);
             }
         }
-            });
-        }
     }
 }

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

@@ -33,11 +33,12 @@ public class RedisGpsMsgListener implements MessageListener {
     @Autowired
     private IStreamPushService streamPushService;
 
-    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
 
 
     @Override
     public void onMessage(@NotNull Message message, byte[] bytes) {
+        log.debug("[REDIS: GPS]： {}", new String(message.getBody()));
         taskQueue.offer(message);
     }
 

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

@@ -8,11 +8,9 @@ import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
 import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
 import com.genersoft.iot.vmp.utils.DateUtil;
 import lombok.extern.slf4j.Slf4j;
-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.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
@@ -38,30 +36,38 @@ public class RedisPushStreamListMsgListener implements MessageListener {
     @Resource
     private IStreamPushService streamPushService;
 
-    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
-
-    @Qualifier("taskExecutor")
-    @Autowired
-    private ThreadPoolTaskExecutor taskExecutor;
 
     @Override
     public void onMessage(Message message, byte[] bytes) {
-        log.info("[REDIS消息-推流设备列表更新]： {}", new String(message.getBody()));
+        log.info("[REDIS: 流设备列表更新]： {}", new String(message.getBody()));
-        boolean isEmpty = taskQueue.isEmpty();
         taskQueue.offer(message);
-        if (isEmpty) {
+    }
-            taskExecutor.execute(() -> {
+
-                while (!taskQueue.isEmpty()) {
+    @Scheduled(fixedDelay = 100)
+    public void executeTaskQueue() {
+        if (taskQueue.isEmpty()) {
+            return;
+        }
+        List<Message> messageDataList = new ArrayList<>();
+        int size = taskQueue.size();
+        for (int i = 0; i < size; i++) {
             Message msg = taskQueue.poll();
+            if (msg != null) {
+                messageDataList.add(msg);
+            }
+        }
+        if (messageDataList.isEmpty()) {
+            return;
+        }
+        for (Message msg : messageDataList) {
             try {
                 List<RedisPushStreamMessage> streamPushItems = JSON.parseArray(new String(msg.getBody()), RedisPushStreamMessage.class);
                 //查询全部的app+stream 用于判断是添加还是修改
                 Map<String, StreamPush> allAppAndStream = streamPushService.getAllAppAndStreamMap();
                 Map<String, StreamPush> allGBId = streamPushService.getAllGBId();
 
-                        /**
+                // 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表
-                         * 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表
-                         */
                 List<StreamPush> streamPushItemForSave = new ArrayList<>();
                 List<StreamPush> streamPushItemForUpdate = new ArrayList<>();
                 for (RedisPushStreamMessage pushStreamMessage : streamPushItems) {
@@ -96,28 +102,27 @@ public class RedisPushStreamListMsgListener implements MessageListener {
                         streamPush.setUpdateTime(DateUtil.getNow());
                         streamPush.setGbDeviceId(pushStreamMessage.getGbId());
                         streamPush.setGbName(pushStreamMessage.getName());
-                                streamPush.setGbStatus(pushStreamMessage.isStatus()?"ON":"OFF");
+                        streamPush.setGbStatus(pushStreamMessage.isStatus() ? "ON" : "OFF");
                         //存在就只修改 name和gbId
                         streamPushItemForUpdate.add(streamPush);
                     }
                 }
                 if (!streamPushItemForSave.isEmpty()) {
-                            log.info("添加{}条",streamPushItemForSave.size());
+                    log.info("添加{}条", streamPushItemForSave.size());
                     log.info(JSONObject.toJSONString(streamPushItemForSave));
                     streamPushService.batchAdd(streamPushItemForSave);
 
                 }
-                        if(!streamPushItemForUpdate.isEmpty()){
+                if (!streamPushItemForUpdate.isEmpty()) {
-                            log.info("修改{}条",streamPushItemForUpdate.size());
+                    log.info("修改{}条", streamPushItemForUpdate.size());
                     log.info(JSONObject.toJSONString(streamPushItemForUpdate));
                     streamPushService.batchUpdate(streamPushItemForUpdate);
                 }
-                    }catch (Exception e) {
+            } catch (Exception e) {
-                        log.warn("[REDIS消息-推流设备列表更新] 发现未处理的异常, \r\n{}", new String(message.getBody()));
+                log.warn("[REDIS消息-推流设备列表更新] 发现未处理的异常, \r\n{}", new String(msg.getBody()));
                 log.error("[REDIS消息-推流设备列表更新] 异常内容： ", e);
             }
         }
-            });
+
-        }
     }
 }

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

@@ -7,16 +7,20 @@ 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 org.springframework.util.ObjectUtils;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 /**
  * 接收redis返回的推流结果
+ *
  * @author lin
  * PUBLISH VM_MSG_STREAM_PUSH_RESPONSE '{"code":0,"msg":"失败","app":"1000","stream":"10000022"}'
  */
@@ -30,24 +34,38 @@ public class RedisPushStreamResponseListener implements MessageListener {
     @Autowired
     private ThreadPoolTaskExecutor taskExecutor;
 
-    private Map<String, PushStreamResponseEvent> responseEvents = new ConcurrentHashMap<>();
+    private final Map<String, PushStreamResponseEvent> responseEvents = new ConcurrentHashMap<>();
 
-    public interface PushStreamResponseEvent{
+    public interface PushStreamResponseEvent {
         void run(MessageForPushChannelResponse response);
     }
 
     @Override
     public void onMessage(Message message, byte[] bytes) {
-        log.info("[REDIS消息-请求推流结果]： {}", new String(message.getBody()));
+        log.info("[REDIS: 推流结果]： {}", new String(message.getBody()));
-        boolean isEmpty = taskQueue.isEmpty();
         taskQueue.offer(message);
-        if (isEmpty) {
+    }
-            taskExecutor.execute(() -> {
+
-                while (!taskQueue.isEmpty()) {
+    @Scheduled(fixedDelay = 100)
+    public void executeTaskQueue() {
+        if (taskQueue.isEmpty()) {
+            return;
+        }
+        List<Message> messageDataList = new ArrayList<>();
+        int size = taskQueue.size();
+        for (int i = 0; i < size; i++) {
             Message msg = taskQueue.poll();
+            if (msg != null) {
+                messageDataList.add(msg);
+            }
+        }
+        if (messageDataList.isEmpty()) {
+            return;
+        }
+        for (Message msg : messageDataList) {
             try {
                 MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
-                        if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
+                if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())) {
                     log.info("[REDIS消息-请求推流结果]：参数不全");
                     continue;
                 }
@@ -55,13 +73,11 @@ public class RedisPushStreamResponseListener implements MessageListener {
                 if (responseEvents.get(response.getApp() + response.getStream()) != null) {
                     responseEvents.get(response.getApp() + response.getStream()).run(response);
                 }
-                    }catch (Exception e) {
+            } catch (Exception e) {
-                        log.warn("[REDIS消息-请求推流结果] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
+                log.warn("[REDIS消息-请求推流结果] 发现未处理的异常, \r\n{}", JSON.toJSONString(msg));
                 log.error("[REDIS消息-请求推流结果] 异常内容： ", e);
             }
         }
-            });
-        }
     }
 
     public void addEvent(String app, String stream, PushStreamResponseEvent callback) {

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

@@ -9,19 +9,21 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
 import lombok.extern.slf4j.Slf4j;
 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.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 
 /**
  * 接收redis发送的推流设备上线下线通知
+ *
  * @author lin
  * 发送 PUBLISH VM_MSG_PUSH_STREAM_STATUS_CHANGE '{"setAllOffline":false,"offlineStreams":[{"app":"1000","stream":"10000022","timeStamp":1726729716551}]}'
  * 订阅 SUBSCRIBE VM_MSG_PUSH_STREAM_STATUS_CHANGE
@@ -44,20 +46,29 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
 
     private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
 
-    @Qualifier("taskExecutor")
-    @Autowired
-    private ThreadPoolTaskExecutor taskExecutor;
-
     @Override
     public void onMessage(Message message, byte[] bytes) {
-        boolean isEmpty = taskQueue.isEmpty();
+        log.info("[REDIS: 流设备状态变化]： {}", new String(message.getBody()));
-        log.warn("[REDIS消息-推流设备状态变化]： {}", new String(message.getBody()));
         taskQueue.offer(message);
+    }
 
-        if (isEmpty) {
+    @Scheduled(fixedDelay = 100)
-            taskExecutor.execute(() -> {
+    public void executeTaskQueue() {
-                while (!taskQueue.isEmpty()) {
+        if (taskQueue.isEmpty()) {
+            return;
+        }
+        List<Message> messageDataList = new ArrayList<>();
+        int size = taskQueue.size();
+        for (int i = 0; i < size; i++) {
             Message msg = taskQueue.poll();
+            if (msg != null) {
+                messageDataList.add(msg);
+            }
+        }
+        if (messageDataList.isEmpty()) {
+            return;
+        }
+        for (Message msg : messageDataList) {
             try {
                 PushStreamStatusChangeFromRedisDto streamStatusMessage = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class);
                 if (streamStatusMessage == null) {
@@ -80,13 +91,11 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
                     // 更新部分设备上线
                     streamPushService.online(streamStatusMessage.getOnlineStreams());
                 }
-                    }catch (Exception e) {
+            } catch (Exception e) {
-                        log.warn("[REDIS消息-推流设备状态变化] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
+                log.warn("[REDIS消息-推流设备状态变化] 发现未处理的异常, \r\n{}", JSON.toJSONString(msg));
                 log.error("[REDIS消息-推流设备状态变化] 异常内容： ", e);
             }
         }
-            });
-        }
     }
 
     @Override
@@ -94,7 +103,7 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
         if (!userSetting.isUsePushingAsStatus()) {
             //  启动时设置所有推流通道离线，发起查询请求
             redisCatchStorage.sendStreamPushRequestedMsgForStatus();
-            dynamicTask.startDelay(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED, ()->{
+            dynamicTask.startDelay(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED, () -> {
                 log.info("[REDIS消息]未收到redis回复推流设备状态，执行推流设备离线");
                 // 五秒收不到请求就设置通道离线，然后通知上级离线
                 streamPushService.allOffline();