优化notify消息处理

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

@@ -1,6 +1,5 @@
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
 
-import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.SipConfig;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.gb28181.bean.Device;
@@ -42,8 +41,8 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 
     private final static Logger logger = LoggerFactory.getLogger(NotifyRequestForCatalogProcessor.class);
 
-	private final List<DeviceChannel> updateChannelOnlineList = new CopyOnWriteArrayList<>();
+	private final List<DeviceChannel> updateChannelForStatusChange = new CopyOnWriteArrayList<>();
-	private final List<DeviceChannel> updateChannelOfflineList = new CopyOnWriteArrayList<>();
+//	private final List<DeviceChannel> updateChannelOfflineList = new CopyOnWriteArrayList<>();
 	private final Map<String, DeviceChannel> updateChannelMap = new ConcurrentHashMap<>();
 
 	private final Map<String, DeviceChannel> addChannelMap = new ConcurrentHashMap<>();
@@ -63,9 +62,6 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 	@Autowired
 	private IDeviceChannelService deviceChannelService;
 
-	@Autowired
-	private DynamicTask dynamicTask;
-
 	@Autowired
 	private SipConfig sipConfig;
 
@@ -77,13 +73,24 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 		}
 		taskQueue.offer(new HandlerCatchData(evt, null, null));
 	}
+//	@Scheduled(fixedRate = 2000)   //每400毫秒执行一次
+//	public void showSize(){
+//		logger.warn("[notify-目录订阅] 待处理消息数量： {}", taskQueue.size() );
+//	}
 
 	@Scheduled(fixedRate = 400)   //每400毫秒执行一次
 	public void executeTaskQueue(){
 		if (taskQueue.isEmpty()) {
 			return;
 		}
-		for (HandlerCatchData take : taskQueue) {
+		List<HandlerCatchData> handlerCatchDataList = new ArrayList<>();
+		while (!taskQueue.isEmpty()) {
+			handlerCatchDataList.add(taskQueue.poll());
+		}
+		if (handlerCatchDataList.isEmpty()) {
+			return;
+		}
+		for (HandlerCatchData take : handlerCatchDataList) {
 			if (take == null) {
 				continue;
 			}
@@ -130,12 +137,15 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 							channel.setParentId(null);
 						}
 						channel.setDeviceId(device.getDeviceId());
-						logger.info("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
+						if (logger.isDebugEnabled()){
+							logger.debug("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
+						}
 						switch (event) {
 							case CatalogEvent.ON:
 								// 上线
 								logger.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
-								updateChannelOnlineList.add(channel);
+								channel.setStatus(true);
+								updateChannelForStatusChange.add(channel);
 								if (userSetting.getDeviceStatusNotify()) {
 									// 发送redis消息
 									redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), true);
@@ -147,7 +157,8 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 								if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
 									logger.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
 								} else {
-									updateChannelOfflineList.add(channel);
+									channel.setStatus(false);
+									updateChannelForStatusChange.add(channel);
 									if (userSetting.getDeviceStatusNotify()) {
 										// 发送redis消息
 										redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
@@ -160,7 +171,8 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 								if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
 									logger.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
 								} else {
-									updateChannelOfflineList.add(channel);
+									channel.setStatus(false);
+									updateChannelForStatusChange.add(channel);
 									if (userSetting.getDeviceStatusNotify()) {
 										// 发送redis消息
 										redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
@@ -173,7 +185,8 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 								if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
 									logger.info("[收到通道视频故障通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
 								} else {
-									updateChannelOfflineList.add(channel);
+									channel.setStatus(false);
+									updateChannelForStatusChange.add(channel);
 									if (userSetting.getDeviceStatusNotify()) {
 										// 发送redis消息
 										redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
@@ -211,6 +224,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 							case CatalogEvent.UPDATE:
 								// 更新
 								logger.info("[收到更新通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
+
 								// 判断此通道是否存在
 								DeviceChannel deviceChannelForUpdate = deviceChannelService.getOne(deviceId, channel.getChannelId());
 								if (deviceChannelForUpdate != null) {
@@ -242,8 +256,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 		taskQueue.clear();
 		if (!updateChannelMap.keySet().isEmpty()
 				|| !addChannelMap.keySet().isEmpty()
-				|| !updateChannelOnlineList.isEmpty()
+				|| !updateChannelForStatusChange.isEmpty()
-				|| !updateChannelOfflineList.isEmpty()
 				|| !deleteChannelList.isEmpty()) {
 			executeSave();
 		}
@@ -256,14 +269,9 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 			logger.error("[存储收到的增加通道] 异常： ", e );
 		}
 		try {
-			executeSaveForOnline();
+			executeSaveForStatus();
 		} catch (Exception e) {
-			logger.error("[存储收到的通道上线] 异常： ", e );
+			logger.error("[存储收到的通道状态变化] 异常： ", e );
-		}
-		try {
-			executeSaveForOffline();
-		} catch (Exception e) {
-			logger.error("[存储收到的通道离线] 异常： ", e );
 		}
 		try {
 			executeSaveForUpdate();
@@ -301,22 +309,10 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
 		}
 	}
 
-	private void executeSaveForOnline(){
+	private void executeSaveForStatus(){
-		if (!updateChannelOnlineList.isEmpty()) {
+		if (!updateChannelForStatusChange.isEmpty()) {
-			deviceChannelService.channelsOnline(updateChannelOnlineList);
+			deviceChannelService.updateChannelsStaus(updateChannelForStatusChange);
-			updateChannelOnlineList.clear();
+			updateChannelForStatusChange.clear();
 		}
 	}
-
-	private void executeSaveForOffline(){
-		if (!updateChannelOfflineList.isEmpty()) {
-			deviceChannelService.channelsOffline(updateChannelOfflineList);
-			updateChannelOfflineList.clear();
-		}
-	}
-
-//	@Scheduled(fixedRate = 10000)   //每1秒执行一次
-//	public void execute(){
-//		logger.info("[待处理Notify-目录订阅消息数量]: {}", taskQueue.size());
-//	}
 }

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

@@ -25,6 +25,7 @@ import org.springframework.util.ObjectUtils;
 
 import javax.sip.RequestEvent;
 import javax.sip.header.FromHeader;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
@@ -68,7 +69,14 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
 		if (taskQueue.isEmpty()) {
 			return;
 		}
-		for (HandlerCatchData take : taskQueue) {
+		List<HandlerCatchData> handlerCatchDataList = new ArrayList<>();
+		while (!taskQueue.isEmpty()) {
+			handlerCatchDataList.add(taskQueue.poll());
+		}
+		if (handlerCatchDataList.isEmpty()) {
+			return;
+		}
+		for (HandlerCatchData take : handlerCatchDataList) {
 			if (take == null) {
 				continue;
 			}

src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java

@@ -67,12 +67,7 @@ public interface IDeviceChannelService {
     /**
      * 批量上线
      */
-    int channelsOnline(List<DeviceChannel> channels);
+    int updateChannelsStaus(List<DeviceChannel> channels);
-
-    /**
-     * 批量下线
-     */
-    int channelsOffline(List<DeviceChannel> channels);
 
     /**
      *  获取一个通道

src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java

@@ -239,13 +239,45 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
     }
 
     @Override
-    public int deleteChannels(List<DeviceChannel> deleteChannelList) {
+    @Transactional
-       return channelMapper.batchDel(deleteChannelList);
+    public int deleteChannels(List<DeviceChannel> channels) {
+        int limitCount = 1000;
+        int result = 0;
+        if (!channels.isEmpty()) {
+            if (channels.size() > limitCount) {
+                for (int i = 0; i < channels.size(); i += limitCount) {
+                    int toIndex = i + limitCount;
+                    if (i + limitCount > channels.size()) {
+                        toIndex = channels.size();
+                    }
+                    result += channelMapper.batchDel(channels.subList(i, toIndex));
+                }
+            }else {
+                result += channelMapper.batchDel(channels);
+            }
+        }
+       return result;
     }
 
     @Override
-    public int channelsOnline(List<DeviceChannel> channels) {
+    @Transactional
-        return channelMapper.batchOnline(channels);
+    public int updateChannelsStaus(List<DeviceChannel> channels) {
+        int limitCount = 1000;
+        int result = 0;
+        if (!channels.isEmpty()) {
+            if (channels.size() > limitCount) {
+                for (int i = 0; i < channels.size(); i += limitCount) {
+                    int toIndex = i + limitCount;
+                    if (i + limitCount > channels.size()) {
+                        toIndex = channels.size();
+                    }
+                    result += channelMapper.batchUpdateStatus(channels.subList(i, toIndex));
+                }
+            }else {
+                result += channelMapper.batchUpdateStatus(channels);
+            }
+        }
+        return result;
     }
 
     @Override
@@ -253,12 +285,6 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
         channelMapper.online(channel.getDeviceId(), channel.getChannelId());
     }
 
-    @Override
-    public int channelsOffline(List<DeviceChannel> channels) {
-        return channelMapper.batchOffline(channels);
-    }
-
-
     @Override
     public void offline(DeviceChannel channel) {
         channelMapper.offline(channel.getDeviceId(), channel.getChannelId());
@@ -275,6 +301,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
     }
 
     @Override
+    @Transactional
     public synchronized void batchUpdateChannel(List<DeviceChannel> channels) {
         String now = DateUtil.getNow();
         for (DeviceChannel channel : channels) {
@@ -297,8 +324,27 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
     }
 
     @Override
+    @Transactional
     public void batchAddChannel(List<DeviceChannel> channels) {
-        channelMapper.batchAdd(channels);
+        String now = DateUtil.getNow();
+        for (DeviceChannel channel : channels) {
+            channel.setUpdateTime(now);
+            channel.setCreateTime(now);
+        }
+        int limitCount = 1000;
+        if (!channels.isEmpty()) {
+            if (channels.size() > limitCount) {
+                for (int i = 0; i < channels.size(); i += limitCount) {
+                    int toIndex = i + limitCount;
+                    if (i + limitCount > channels.size()) {
+                        toIndex = channels.size();
+                    }
+                    channelMapper.batchAdd(channels.subList(i, toIndex));
+                }
+            }else {
+                channelMapper.batchAdd(channels);
+            }
+        }
         for (DeviceChannel channel : channels) {
             if (channel.getParentId() != null) {
                 channelMapper.updateChannelSubCount(channel.getDeviceId(), channel.getParentId());

src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java

@@ -479,10 +479,10 @@ public interface DeviceChannelMapper {
 
     @Update({"<script>" +
             "<foreach collection='channels' item='item' separator=';'>" +
-            "UPDATE wvp_device_channel SET status=true WHERE device_id=#{item.deviceId} AND channel_id=#{item.channelId}" +
+            "UPDATE wvp_device_channel SET status=#{item.status} WHERE device_id=#{item.deviceId} AND channel_id=#{item.channelId}" +
             "</foreach>" +
             "</script>"})
-    int batchOnline(@Param("channels") List<DeviceChannel> channels);
+    int batchUpdateStatus(@Param("channels") List<DeviceChannel> channels);
 
     @Update({"<script>" +
             "<foreach collection='channels' item='item' separator=';'>" +