From 5bcd8495e03b08ff417f74f816d20e028f9b1fe0 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期日, 20 二月 2022 16:22:43 +0800 Subject: [PATCH] 兼容海康不规范的xml,大幅度提高通道分页查询速度,优化节点的保活。 --- src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 128 ++++++++++++++++++++++++++++++++++++++---- 1 files changed, 115 insertions(+), 13 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java index 3fac37a..ff31d8a 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java @@ -5,19 +5,22 @@ import com.alibaba.fastjson.TypeReference; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.UserSetup; +import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; import com.genersoft.iot.vmp.gb28181.bean.GbStream; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; +import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog; +import com.genersoft.iot.vmp.gb28181.event.EventPublisher; +import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig; import com.genersoft.iot.vmp.media.zlm.dto.*; +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.storager.IRedisCatchStorage; -import com.genersoft.iot.vmp.storager.dao.GbStreamMapper; -import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper; -import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper; -import com.genersoft.iot.vmp.storager.dao.StreamPushMapper; +import com.genersoft.iot.vmp.storager.dao.*; +import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto; import com.github.pagehelper.PageHelper; import com.github.pagehelper.PageInfo; import org.springframework.beans.factory.annotation.Autowired; @@ -39,7 +42,16 @@ private ParentPlatformMapper parentPlatformMapper; @Autowired + private PlatformCatalogMapper platformCatalogMapper; + + @Autowired private PlatformGbStreamMapper platformGbStreamMapper; + + @Autowired + private IGbStreamService gbStreamService; + + @Autowired + private EventPublisher eventPublisher; @Autowired private ZLMRESTfulUtils zlmresTfulUtils; @@ -84,13 +96,12 @@ streamPushItem.setMediaServerId(item.getMediaServerId()); streamPushItem.setStream(item.getStream()); streamPushItem.setAliveSecond(item.getAliveSecond()); - streamPushItem.setCreateStamp(item.getCreateStamp()); streamPushItem.setOriginSock(item.getOriginSock()); streamPushItem.setTotalReaderCount(item.getTotalReaderCount()); streamPushItem.setOriginType(item.getOriginType()); streamPushItem.setOriginTypeStr(item.getOriginTypeStr()); streamPushItem.setOriginUrl(item.getOriginUrl()); - streamPushItem.setCreateStamp(item.getCreateStamp()); + streamPushItem.setCreateStamp(item.getCreateStamp() * 1000); streamPushItem.setAliveSecond(item.getAliveSecond()); streamPushItem.setStatus(true); streamPushItem.setStreamType("push"); @@ -99,22 +110,24 @@ } @Override - public PageInfo<StreamPushItem> getPushList(Integer page, Integer count) { + public PageInfo<StreamPushItem> getPushList(Integer page, Integer count, String query, Boolean pushing, String mediaServerId) { PageHelper.startPage(page, count); - List<StreamPushItem> all = streamPushMapper.selectAll(); + List<StreamPushItem> all = streamPushMapper.selectAllForList(query, pushing, mediaServerId); return new PageInfo<>(all); } @Override public List<StreamPushItem> getPushList(String mediaServerId) { - return streamPushMapper.selectAllByMediaServerId(mediaServerId); + return streamPushMapper.selectAllByMediaServerIdWithOutGbID(mediaServerId); } @Override public boolean saveToGB(GbStream stream) { stream.setStreamType("push"); stream.setStatus(true); + stream.setCreateStamp(System.currentTimeMillis()); int add = gbStreamMapper.add(stream); + // 鏌ユ壘寮�鍚簡鍏ㄩ儴鐩存挱娴佸叡浜殑涓婄骇骞冲彴 List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream(); if (parentPlatforms.size() > 0) { @@ -122,18 +135,30 @@ stream.setCatalogId(parentPlatform.getCatalogId()); stream.setPlatformId(parentPlatform.getServerGBId()); String streamId = stream.getStream(); - StreamProxyItem streamProxyItems = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId()); - if (streamProxyItems == null) { + StreamProxyItem streamProxyItem = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId()); + if (streamProxyItem == null) { platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + }else { + if (!streamProxyItem.getGbId().equals(stream.getGbId())) { + // 姝ゆ祦浣跨敤鍙︿竴涓浗鏍嘔d宸茬粡涓庤骞冲彴鍏宠仈锛岀Щ闄ゆ璁板綍 + platformGbStreamMapper.delByAppAndStreamAndPlatform(stream.getApp(), streamId, parentPlatform.getServerGBId()); + platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + } } } } + return add > 0; } @Override public boolean removeFromGB(GbStream stream) { + // 鍒ゆ柇鏄惁闇�瑕佸彂閫佷簨浠� + gbStreamService.sendCatalogMsg(stream, CatalogEvent.DEL); int del = gbStreamMapper.del(stream.getApp(), stream.getStream()); + platformGbStreamMapper.delByAppAndStream(stream.getApp(), stream.getStream()); MediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId()); JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, stream.getApp(), stream.getStream()); if (mediaList == null) { @@ -152,6 +177,8 @@ @Override public boolean stop(String app, String streamId) { StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId); + gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL); + int delStream = streamPushMapper.del(app, streamId); gbStreamMapper.del(app, streamId); platformGbStreamMapper.delByAppAndStream(app, streamId); @@ -177,7 +204,9 @@ Map<String, MediaItem> streamInfoPushItemMap = new HashMap<>(); if (pushList.size() > 0) { for (StreamPushItem streamPushItem : pushList) { - pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem); + if (StringUtils.isEmpty(streamPushItem.getGbId())) { + pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem); + } } } if (mediaItems.size() > 0) { @@ -241,7 +270,7 @@ @Override public void zlmServerOffline(String mediaServerId) { - List<StreamPushItem> streamPushItems = streamPushMapper.selectAllByMediaServerId(mediaServerId); + List<StreamPushItem> streamPushItems = streamPushMapper.selectAllByMediaServerIdWithOutGbID(mediaServerId); // 绉婚櫎娌℃湁GBId鐨勬帹娴� streamPushMapper.deleteWithoutGBId(mediaServerId); gbStreamMapper.deleteWithoutGBId("push", mediaServerId); @@ -279,6 +308,7 @@ streamPushItem.setStreamType("push"); streamPushItem.setStatus(true); streamPushItem.setGbId("34020000004111" + gbId); + streamPushItem.setCreateStamp(System.currentTimeMillis()); gbId ++; } int limitCount = 30; @@ -296,4 +326,76 @@ } return true; } + + @Override + public void batchAdd(List<StreamPushItem> streamPushItems) { + streamPushMapper.addAll(streamPushItems); + gbStreamMapper.batchAdd(streamPushItems); + // 鏌ユ壘寮�鍚簡鍏ㄩ儴鐩存挱娴佸叡浜殑涓婄骇骞冲彴 + List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream(); + if (parentPlatforms.size() > 0) { + for (StreamPushItem stream : streamPushItems) { + for (ParentPlatform parentPlatform : parentPlatforms) { + stream.setCatalogId(parentPlatform.getCatalogId()); + stream.setPlatformId(parentPlatform.getServerGBId()); + String streamId = stream.getStream(); + StreamProxyItem streamProxyItem = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId()); + if (streamProxyItem == null) { + platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + }else { + if (!streamProxyItem.getGbId().equals(stream.getGbId())) { + // 姝ゆ祦浣跨敤鍙︿竴涓浗鏍嘔d宸茬粡涓庤骞冲彴鍏宠仈锛岀Щ闄ゆ璁板綍 + platformGbStreamMapper.delByAppAndStreamAndPlatform(stream.getApp(), streamId, parentPlatform.getServerGBId()); + platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + stream.setGbId(streamProxyItem.getGbId()); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.DEL); + } + } + } + } + } + } + + @Override + public void batchAddForUpload(String platformId, String catalogId, List<StreamPushItem> streamPushItems) { + streamPushMapper.addAll(streamPushItems); + gbStreamMapper.batchAdd(streamPushItems); + if (platformId != null) { + ParentPlatform platform = parentPlatformMapper.getParentPlatByServerGBId(platformId); + if (platform != null) { + if (catalogId == null) { + catalogId = platform.getCatalogId(); + }else { + PlatformCatalog catalog = platformCatalogMapper.select(catalogId); + if (catalog == null) { + return; + } + } + platformGbStreamMapper.batchAdd(platformId, catalogId, streamPushItems); + eventPublisher.catalogEventPublishForStream(platformId, streamPushItems.toArray(new GbStream[0]), CatalogEvent.ADD); + } + } + } + + @Override + public boolean batchStop(List<GbStream> gbStreams) { + if (gbStreams == null || gbStreams.size() == 0) { + return false; + } + gbStreamService.sendCatalogMsgs(gbStreams, CatalogEvent.DEL); + + int delStream = streamPushMapper.delAllForGbStream(gbStreams); + gbStreamMapper.batchDelForGbStream(gbStreams); + platformGbStreamMapper.delByGbStreams(gbStreams); + if (delStream > 0) { + for (GbStream gbStream : gbStreams) { + MediaServerItem mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId()); + zlmresTfulUtils.closeStreams(mediaServerItem, gbStream.getApp(), gbStream.getStream()); + } + + } + return true; + } } -- Gitblit v1.8.0