Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化队列的处理逻辑

Browse Source
This commit is contained in:
648540858 2022-08-17 15:38:06 +08:00
parent d7865d55e1
commit aa7ab1e479
4 changed files with 86 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -116,14 +116,17 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
logger.info("接收到消息：" + cmd); logger.info("接收到消息：" + cmd);
} }
} catch (DocumentException e) { } catch (DocumentException e) {
throw new RuntimeException(e); logger.error("处理NOTIFY消息时错误", e);
} finally {
taskQueueHandlerRun = false;
} }
} }
taskQueueHandlerRun = false;
}); });
} }
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace(); e.printStackTrace();
} finally {
taskQueueHandlerRun = false;
} }
} }

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

@ -35,6 +35,9 @@ import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentLinkedQueue;
/**
* 目录查询的回复
*/
@Component @Component
public class CatalogResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler { public class CatalogResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler {
@ -85,82 +88,83 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
// 回复200 OK // 回复200 OK
try { try {
responseAck(evt, Response.OK); responseAck(evt, Response.OK);
if (!taskQueueHandlerRun) {
taskQueueHandlerRun = true;
taskExecutor.execute(()-> {
while (!taskQueue.isEmpty()) {
HandlerCatchData take = taskQueue.poll();
try {
Element rootElement = getRootElement(take.getEvt(), take.getDevice().getCharset());
Element deviceListElement = rootElement.element("DeviceList");
Element sumNumElement = rootElement.element("SumNum");
Element snElement = rootElement.element("SN");
if (snElement == null || sumNumElement == null || deviceListElement == null) {
responseAck(take.getEvt(), Response.BAD_REQUEST, "xml error");
return;
}
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 (DocumentException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
} catch (SipException e) {
e.printStackTrace();
} finally {
taskQueueHandlerRun = false;
}
}
});
}
} catch (SipException e) { } catch (SipException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} catch (InvalidArgumentException e) { } catch (InvalidArgumentException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} catch (ParseException e) { } catch (ParseException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} finally {
taskQueueHandlerRun = false;
} }
if (!taskQueueHandlerRun) {
taskQueueHandlerRun = true;
taskExecutor.execute(()-> {
while (!taskQueue.isEmpty()) {
HandlerCatchData take = taskQueue.poll();
String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + take.getDevice().getDeviceId();
Element rootElement = null;
try {
rootElement = getRootElement(take.getEvt(), take.getDevice().getCharset());
Element deviceListElement = rootElement.element("DeviceList");
Element sumNumElement = rootElement.element("SumNum");
Element snElement = rootElement.element("SN");
if (snElement == null || sumNumElement == null || deviceListElement == null) {
responseAck(take.getEvt(), Response.BAD_REQUEST, "xml error");
return;
}
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 (DocumentException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
} catch (SipException e) {
e.printStackTrace();
}
}
taskQueueHandlerRun = false;
});
}
} }
@Override @Override

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

@ -76,8 +76,8 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
if (!taskQueueHandlerRun) { if (!taskQueueHandlerRun) {
taskQueueHandlerRun = true; taskQueueHandlerRun = true;
taskExecutor.execute(()->{ taskExecutor.execute(()->{
try { while (!taskQueue.isEmpty()) {
while (!taskQueue.isEmpty()) { try {
HandlerCatchData take = taskQueue.poll(); HandlerCatchData take = taskQueue.poll();
Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset()); Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset());
String sn = getText(rootElementForCharset, "SN"); String sn = getText(rootElementForCharset, "SN");
@ -141,10 +141,11 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
releaseRequest(take.getDevice().getDeviceId(), sn); releaseRequest(take.getDevice().getDeviceId(), sn);
} }
} }
} catch (DocumentException e) {
throw new RuntimeException(e);
} finally {
taskQueueHandlerRun = false;
} }
taskQueueHandlerRun = false;
}catch (DocumentException e) {
throw new RuntimeException(e);
} }
}); });
} }
@ -155,6 +156,8 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
e.printStackTrace(); e.printStackTrace();
} catch (ParseException e) { } catch (ParseException e) {
e.printStackTrace(); e.printStackTrace();
}finally {
taskQueueHandlerRun = false;
} }
} }

1
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
View File

@ -105,6 +105,7 @@ public class DeviceServiceImpl implements IDeviceService {
redisCatchStorage.updateDevice(device); redisCatchStorage.updateDevice(device);
commander.deviceInfoQuery(device); commander.deviceInfoQuery(device);
sync(device); sync(device);
// TODO 如果设备下的通道级联到了其他平台，那么需要发送事件或者notify给上级平台
}else { }else {
deviceMapper.update(device); deviceMapper.update(device);
redisCatchStorage.updateDevice(device); redisCatchStorage.updateDevice(device);