diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java
index 155d5f73..5e2261ca 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java
+++ b/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> updateChannelOfflineList = new CopyOnWriteArrayList<>();
+ private final List<DeviceChannel> updateChannelForStatusChange = 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()
- || !updateChannelOfflineList.isEmpty()
+ || !updateChannelForStatusChange.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 );
- }
- try {
- executeSaveForOffline();
- } catch (Exception e) {
- logger.error("[存储收到的通道离线] 异常： ", e );
+ logger.error("[存储收到的通道状态变化] 异常： ", e );
}
try {
executeSaveForUpdate();
@@ -301,22 +309,10 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
}
}
- private void executeSaveForOnline(){
- if (!updateChannelOnlineList.isEmpty()) {
- deviceChannelService.channelsOnline(updateChannelOnlineList);
- updateChannelOnlineList.clear();
+ private void executeSaveForStatus(){
+ if (!updateChannelForStatusChange.isEmpty()) {
+ deviceChannelService.updateChannelsStaus(updateChannelForStatusChange);
+ 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());
-// }
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java
index 75a3b0d2..9d24f3ea 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java
+++ b/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;
}
diff --git a/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java b/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
index a8cd5813..39b5939a 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
@@ -67,12 +67,7 @@ public interface IDeviceChannelService {
/**
* 批量上线
*/
- int channelsOnline(List<DeviceChannel> channels);
-
- /**
- * 批量下线
- */
- int channelsOffline(List<DeviceChannel> channels);
+ int updateChannelsStaus(List<DeviceChannel> channels);
/**
* 获取一个通道
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
index 58e7b02a..a82d7ebd 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
+++ b/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) {
- return channelMapper.batchDel(deleteChannelList);
+ @Transactional
+ 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) {
- return channelMapper.batchOnline(channels);
+ @Transactional
+ 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());
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
index 15552211..0172be2b 100755
--- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
+++ b/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=';'>" +