roouzenn/wvp
Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

规范数据库，添加必要约束，优化通道批量导入功能

Browse Source
This commit is contained in:
648540858
2022-02-24 16:55:06 +08:00
parent 2157bb0270
commit a42dda2bd3
13 changed files with 205 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
View File

@@ -9,6 +9,7 @@ import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto;
import com.github.pagehelper.PageInfo;
import java.util.List;
import java.util.Map;
public interface IStreamPushService {
@@ -69,5 +70,5 @@ public interface IStreamPushService {
boolean batchStop(List<GbStream> streamPushItems);
void batchAddForUpload(String platformId, String catalogId, List<StreamPushItem> streamPushItems);
void batchAddForUpload(List<StreamPushItem> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll);
}

4
src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java
View File

@@ -159,7 +159,9 @@ public class GbStreamServiceImpl implements IGbStreamService {
List<ParentPlatform> parentPlatforms = platformGbStreamMapper.selectByAppAndStream(gs.getApp(), gs.getStream());
if (parentPlatforms.size() > 0) {
for (ParentPlatform parentPlatform : parentPlatforms) {
eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), gs, type);
if (parentPlatform != null) {
eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), gs, type);
}
}
}
}

69
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
View File

@@ -28,6 +28,7 @@ import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.util.*;
import java.util.stream.Collectors;
@Service
public class StreamPushServiceImpl implements IStreamPushService {
@@ -359,23 +360,63 @@ public class StreamPushServiceImpl implements IStreamPushService {
}
@Override
public void batchAddForUpload(String platformId, String catalogId, List<StreamPushItem> streamPushItems) {
public void batchAddForUpload(List<StreamPushItem> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll ) {
// 存储数据到stream_push表
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;
List<StreamPushItem> streamPushItemForGbStream = streamPushItems.stream()
.filter(streamPushItem-> streamPushItem.getId() != null)
.collect(Collectors.toList());
// 存储数据到gb_stream表， id会返回到streamPushItemForGbStream里
if (streamPushItemForGbStream.size() > 0) {
gbStreamMapper.batchAdd(streamPushItemForGbStream);
}
// 去除没有ID也就是没有存储到数据库的数据
List<StreamPushItem> streamPushItemsForPlatform = streamPushItemForGbStream.stream()
.filter(streamPushItem-> streamPushItem.getGbStreamId() != null)
.collect(Collectors.toList());
if (streamPushItemsForPlatform.size() > 0) {
List<StreamPushItem> streamPushItemListFroPlatform = new ArrayList<>();
Map<String, List<StreamPushItem>> platformForEvent = new HashMap<>();
// 遍历存储结果，查找app+Stream->platformId+catalogId的对应关系，然后执行批量写入
for (StreamPushItem streamPushItem : streamPushItemsForPlatform) {
List<String[]> platFormInfoList = streamPushItemsForAll.get(streamPushItem.getApp() + streamPushItem.getStream());
if (platFormInfoList != null) {
if (platFormInfoList.size() > 0) {
for (String[] platFormInfoArray : platFormInfoList) {
StreamPushItem streamPushItemForPlatform = new StreamPushItem();
streamPushItemForPlatform.setGbStreamId(streamPushItem.getGbStreamId());
if (platFormInfoArray.length > 0) {
// 数组 platFormInfoArray 0 为平台ID。 1为目录ID
streamPushItemForPlatform.setPlatformId(platFormInfoArray[0]);
List<StreamPushItem> streamPushItemsInPlatform = platformForEvent.get(streamPushItem.getPlatformId());
if (streamPushItemsInPlatform == null) {
streamPushItemsInPlatform = new ArrayList<>();
platformForEvent.put(platFormInfoArray[0], streamPushItemsInPlatform);
}
// 为发送通知整理数据
streamPushItemForPlatform.setApp(streamPushItem.getApp());
streamPushItemForPlatform.setStream(streamPushItem.getStream());
streamPushItemForPlatform.setGbId(streamPushItem.getGbId());
streamPushItemsInPlatform.add(streamPushItemForPlatform);
}
if (platFormInfoArray.length > 1) {
streamPushItemForPlatform.setCatalogId(platFormInfoArray[1]);
}
streamPushItemListFroPlatform.add(streamPushItemForPlatform);
}
}
}
List<GbStream> gbStreamList = gbStreamMapper.selectAllForAppAndStream(streamPushItems);
platformGbStreamMapper.batchAdd(platformId, catalogId, gbStreamList);
eventPublisher.catalogEventPublishForStream(platformId, streamPushItems.toArray(new GbStream[0]), CatalogEvent.ADD);
}
platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
// 发送通知
for (String platformId : platformForEvent.keySet()) {
eventPublisher.catalogEventPublishForStream(
platformId, platformForEvent.get(platformId).toArray(new GbStream[0]), CatalogEvent.ADD);
}
}
}

121
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushUploadFileHandler.java
View File

@@ -5,21 +5,45 @@ import com.alibaba.excel.event.AnalysisEventListener;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto;
import com.google.common.collect.BiMap;
import com.google.common.collect.HashBiMap;
import org.springframework.util.StringUtils;
import java.util.*;
public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPushExcelDto> {
// 错误数据的回调，用于将错误数据发送给页面
private ErrorDataHandler errorDataHandler;
// 推流的业务类用于存储数据
private IStreamPushService pushService;
// 默认流媒体节点ID
private String defaultMediaServerId;
// 用于存储不加过滤的所有数据
private List<StreamPushItem> streamPushItems = new ArrayList<>();
private Map<String, UploadData> streamPushItemsForPlatform = new HashMap<>();
// 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表
private Map<String,StreamPushItem> streamPushItemForSave = new HashMap<>();
// 用于存储按照APP+Stream为KEY， 平台ID+目录Id 为value的数据，用于存储到gb_stream表后获取app+Stream对应的平台与目录信息，然后存入关联表
private Map<String, List<String[]>> streamPushItemsForPlatform = new HashMap<>();
// 用于判断文件是否存在重复的app+Stream+平台ID
private Set<String> streamPushStreamSet = new HashSet<>();
private Map<String,String> streamPushGBMap = new HashMap<>();
// 用于存储APP+Stream->国标ID 的数据结构, 数据一一对应，全局判断APP+Stream->国标ID是否存在不对应
private BiMap<String,String> gBMap = HashBiMap.create();
// 记录错误的APP+Stream
private List<String> errorStreamList = new ArrayList<>();
// 记录错误的国标ID
private List<String> errorGBList = new ArrayList<>();
// 读取数量计数器
private int loadedSize = 0;
@@ -33,16 +57,6 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPus
void handle(List<String> streams, List<String> gbId);
}
private class UploadData{
public String platformId;
public Map<String, List<StreamPushItem>> catalogData = new HashMap<>();
public List<StreamPushItem> streamPushItems = new ArrayList<>();
public UploadData(String platformId) {
this.platformId = platformId;
}
}
@Override
public void invoke(StreamPushExcelDto streamPushExcelDto, AnalysisContext analysisContext) {
if (StringUtils.isEmpty(streamPushExcelDto.getApp())
@@ -50,18 +64,28 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPus
|| StringUtils.isEmpty(streamPushExcelDto.getGbId())) {
return;
}
if (streamPushGBMap.get(streamPushExcelDto.getApp() + streamPushExcelDto.getStream()) == null) {
streamPushGBMap.put(streamPushExcelDto.getApp() + streamPushExcelDto.getStream(), streamPushExcelDto.getGbId());
if (gBMap.get(streamPushExcelDto.getApp() + streamPushExcelDto.getStream()) == null) {
try {
gBMap.put(streamPushExcelDto.getApp() + streamPushExcelDto.getStream(), streamPushExcelDto.getGbId());
}catch (IllegalArgumentException e) {
e.printStackTrace();
errorGBList.add(streamPushExcelDto.getGbId() + "(不同的app+stream使用了相同的国标ID)");
return;
}
}else {
if (!streamPushGBMap.get(streamPushExcelDto.getApp() + streamPushExcelDto.getStream()).equals(streamPushExcelDto.getGbId())) {
errorGBList.add(streamPushExcelDto.getGbId() + "(同一组app+stream使用了不同国标ID)");
if (!gBMap.get(streamPushExcelDto.getApp() + streamPushExcelDto.getStream()).equals(streamPushExcelDto.getGbId())) {
errorGBList.add(streamPushExcelDto.getGbId() + "(同一组app+stream使用了不同的国标ID)");
return;
}
}
if (streamPushStreamSet.contains(streamPushExcelDto.getApp() + streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId())) {
errorStreamList.add(streamPushExcelDto.getApp() + "/" + streamPushExcelDto.getStream()+ "/" + streamPushExcelDto.getPlatformId() + "(同一组app+stream添加在了同一个平台下)");
errorStreamList.add(streamPushExcelDto.getApp() + "/" + streamPushExcelDto.getStream()+ "/" +
streamPushExcelDto.getPlatformId() + "(同一组app+stream添加在了同一个平台下)");
return;
}else {
streamPushStreamSet.add(streamPushExcelDto.getApp()+streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId());
}
StreamPushItem streamPushItem = new StreamPushItem();
@@ -78,32 +102,31 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPus
streamPushItem.setTotalReaderCount("0");
streamPushItem.setPlatformId(streamPushExcelDto.getPlatformId());
streamPushItem.setCatalogId(streamPushExcelDto.getCatalogId());
if (StringUtils.isEmpty(streamPushExcelDto.getPlatformId())) {
streamPushItems.add(streamPushItem);
}else {
UploadData uploadData = streamPushItemsForPlatform.get(streamPushExcelDto.getPlatformId());
if (uploadData == null) {
uploadData = new UploadData(streamPushExcelDto.getPlatformId());
streamPushItemsForPlatform.put(streamPushExcelDto.getPlatformId(), uploadData);
}
if (!StringUtils.isEmpty(streamPushExcelDto.getCatalogId())) {
List<StreamPushItem> streamPushItems = uploadData.catalogData.get(streamPushExcelDto.getCatalogId());
if (streamPushItems == null) {
streamPushItems = new ArrayList<>();
uploadData.catalogData.put(streamPushExcelDto.getCatalogId(), streamPushItems);
}
streamPushItems.add(streamPushItem);
}else {
uploadData.streamPushItems.add(streamPushItem);
}
// 存入所有的通道信息
streamPushItems.add(streamPushItem);
streamPushItemForSave.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem);
if (!StringUtils.isEmpty(streamPushExcelDto.getPlatformId())) {
List<String[]> platformList = streamPushItemsForPlatform.get(streamPushItem.getApp() + streamPushItem.getStream());
if (platformList == null) {
platformList = new ArrayList<>();
streamPushItemsForPlatform.put(streamPushItem.getApp() + streamPushItem.getStream(), platformList);
}
String platformId = streamPushExcelDto.getPlatformId();
String catalogId = streamPushExcelDto.getCatalogId();
if (StringUtils.isEmpty(streamPushExcelDto.getCatalogId())) {
catalogId = null;
}
String[] platFormInfoArray = new String[]{platformId, catalogId};
platformList.add(platFormInfoArray);
}
streamPushStreamSet.add(streamPushExcelDto.getApp()+streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId());
loadedSize ++;
if (loadedSize > 1000) {
saveData();
streamPushItems.clear();
streamPushItemForSave.clear();
streamPushItemsForPlatform.clear();
loadedSize = 0;
}
@@ -114,30 +137,18 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPus
public void doAfterAllAnalysed(AnalysisContext analysisContext) {
// 这里也要保存数据，确保最后遗留的数据也存储到数据库
saveData();
streamPushGBMap.clear();
streamPushItems.clear();
streamPushItemForSave.clear();
gBMap.clear();
streamPushStreamSet.clear();
streamPushItemsForPlatform.clear();
errorDataHandler.handle(errorStreamList, errorGBList);
}
private void saveData(){
if (streamPushItems.size() > 0) {
pushService.batchAddForUpload(null, null, streamPushItems);
}
// 处理已分配到平台的流
if (streamPushItemsForPlatform.size() > 0){
for (String platformId : streamPushItemsForPlatform.keySet()) {
UploadData uploadData = streamPushItemsForPlatform.get(platformId);
if (uploadData.streamPushItems.size() > 0) {
pushService.batchAddForUpload(platformId, null, uploadData.streamPushItems);
}
if (uploadData.catalogData.size() > 0) {
for (String catalogId : uploadData.catalogData.keySet()) {
if (uploadData.catalogData.get(catalogId).size() > 0) {
pushService.batchAddForUpload(platformId, catalogId, uploadData.catalogData.get(catalogId));
}
}
}
}
if (streamPushItemForSave.size() > 0) {
// 向数据库查询是否存在重复的app
pushService.batchAddForUpload(new ArrayList<>(streamPushItemForSave.values()), streamPushItemsForPlatform);
}
}
}