规范数据库，给各个标设置主键ID

648540858

2022-02-22 2157bb0270663df35b7267c3d10c8b594400102e

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

@@ -5,23 +5,65 @@
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 org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.*;

public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPushExcelDto> {

    private ErrorDataHandler errorDataHandler;
    private IStreamPushService pushService;
    private String defaultMediaServerId;
    private List<StreamPushItem> streamPushItems = new ArrayList<>();
    private Map<String, UploadData> streamPushItemsForPlatform = new HashMap<>();
    private Set<String> streamPushStreamSet = new HashSet<>();
    private Map<String,String> streamPushGBMap = new HashMap<>();
    private List<String> errorStreamList = new ArrayList<>();
    private List<String> errorGBList = new ArrayList<>();
    // 读取数量计数器
    private int loadedSize = 0;

    public StreamPushUploadFileHandler(IStreamPushService pushService, String defaultMediaServerId) {
    public StreamPushUploadFileHandler(IStreamPushService pushService, String defaultMediaServerId, ErrorDataHandler errorDataHandler) {
        this.pushService = pushService;
        this.defaultMediaServerId = defaultMediaServerId;
        this.errorDataHandler = errorDataHandler;
    }

    public interface ErrorDataHandler{
        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())
                || StringUtils.isEmpty(streamPushExcelDto.getStream())
                || StringUtils.isEmpty(streamPushExcelDto.getGbId())) {
            return;
        }
        if (streamPushGBMap.get(streamPushExcelDto.getApp() + streamPushExcelDto.getStream()) == null) {
            streamPushGBMap.put(streamPushExcelDto.getApp() + streamPushExcelDto.getStream(), streamPushExcelDto.getGbId());

        }else {
            if (!streamPushGBMap.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添加在了同一个平台下)");
            return;
        }

        StreamPushItem streamPushItem = new StreamPushItem();
        streamPushItem.setApp(streamPushExcelDto.getApp());
        streamPushItem.setStream(streamPushExcelDto.getStream());
@@ -34,17 +76,68 @@
        streamPushItem.setOriginType(2);
        streamPushItem.setOriginTypeStr("rtsp_push");
        streamPushItem.setTotalReaderCount("0");
        streamPushItems.add(streamPushItem);
        if (streamPushItems.size() > 300) {
            pushService.batchAdd(streamPushItems);
            // 存储完成清理 list
            streamPushItems.clear();
        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);
            }

        }

        streamPushStreamSet.add(streamPushExcelDto.getApp()+streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId());
        loadedSize ++;
        if (loadedSize > 1000) {
            saveData();
            streamPushItems.clear();
            streamPushItemsForPlatform.clear();
            loadedSize = 0;
        }

    }

    @Override
    public void doAfterAllAnalysed(AnalysisContext analysisContext) {
        // 这里也要保存数据，确保最后遗留的数据也存储到数据库
        pushService.batchAdd(streamPushItems);
        saveData();
        streamPushGBMap.clear();
        streamPushStreamSet.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));
                        }
                    }
                }
            }
        }
    }
}