From 64efb660b2c119c00432434c0f651f8996483f18 Mon Sep 17 00:00:00 2001 From: fuliqi <fuliqi@qq.com> Date: 星期五, 21 二月 2025 15:51:13 +0800 Subject: [PATCH] OSD大华数据格式+数据中心重点标签统计数 --- ycl-server/src/main/java/com/ycl/task/UYTask.java | 138 ++++++++++++++++++++++++++++++--------------- 1 files changed, 92 insertions(+), 46 deletions(-) diff --git a/ycl-server/src/main/java/com/ycl/task/UYTask.java b/ycl-server/src/main/java/com/ycl/task/UYTask.java index 2f92fa3..c351a65 100644 --- a/ycl-server/src/main/java/com/ycl/task/UYTask.java +++ b/ycl-server/src/main/java/com/ycl/task/UYTask.java @@ -3,6 +3,7 @@ import com.alibaba.fastjson2.JSONObject; import com.mongodb.client.result.DeleteResult; import com.ycl.feign.UYClient; +import com.ycl.platform.domain.entity.TMonitor; import com.ycl.platform.domain.entity.WorkOrder; import com.ycl.platform.domain.param.UY.ImageDetectionParam; import com.ycl.platform.domain.param.UY.MonitorQualifyParam; @@ -15,6 +16,7 @@ import com.ycl.platform.domain.result.UY.VideoOnlineResult; import com.ycl.platform.domain.vo.UpdateOnlineVO; import com.ycl.platform.mapper.TMonitorMapper; +import com.ycl.platform.mapper.WorkOrderMapper; import com.ycl.platform.service.UYErrorTypeCheckService; import com.ycl.platform.service.WorkOrderService; import com.ycl.platform.service.YwPointService; @@ -24,11 +26,14 @@ import com.ycl.utils.CheckPointUtil; import com.ycl.utils.DateUtils; import constant.ApiConstants; +import constant.CheckConstants; import constant.RedisConstant; import enumeration.ErrorType; import enumeration.general.WorkOrderStatusEnum; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Criteria; @@ -40,6 +45,7 @@ import java.text.SimpleDateFormat; import java.util.*; import java.util.concurrent.*; +import java.util.function.Function; import java.util.stream.Collectors; //浼樹簯瀵规帴鏁版嵁浠诲姟 @@ -54,6 +60,7 @@ private final YwPointService pointService; private final TMonitorMapper monitorMapper; private final WorkOrderService workOrderService; + private final WorkOrderMapper workOrderMapper; private final SysConfigMapper sysConfigMapper; private final CheckPointUtil checkPointUtil; @Value("${youYun.tenantId}") @@ -68,8 +75,8 @@ private String passwd; private static final ExecutorService executorService = new ThreadPoolExecutor(16, - 128, - 5000, + 64, + 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy() @@ -100,7 +107,7 @@ item.setNo(item.getDeviceId()); } }); - pointService.setDeviceTagByGB(records); + pointService.setDeviceTagByGB(records, CheckConstants.Rule_Category_Video); //瀛樻斁鍦╩ongo涓� mongoTemplate.insertAll(records); // 宸ュ崟鐢熸垚 @@ -149,7 +156,7 @@ item.setNo(item.getSerialNumber().getShowValue()); } }); - pointService.setDeviceTagByGB(records); + pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Or); pointService.setNew(records); //瀛樻斁鍦╩ongo涓� mongoTemplate.insertAll(records); @@ -173,8 +180,9 @@ /** * 鐐逛綅鍦ㄧ嚎PING妫�娴� + * 浠诲姟浼氬厛鎵ц涓�娆′紭浜戝悓姝ワ紝鐒跺悗鎵цping妫�娴� + * online瀛楁鏉ヨ嚜浜庝紭浜戯紝pingOnline涓轰富鍔╬ing妫�娴嬬殑銆傚瓨鍏ongo缁欐暟鎹腑蹇冩煡闃� */ - //TODO锛氱偣浣嶅湪绾匡紙淇敼閫昏緫鍙拡瀵瑰伐鍗曪級 public void pointOnline() throws ExecutionException, InterruptedException { log.info("寮�濮嬫娴嬬偣浣嶅湪绾�"); Integer times = 2; @@ -188,6 +196,17 @@ } // 鍏堟煡鍑鸿澶嘔P闆嗗悎,鍓旈櫎鎺夊湪绾挎儏鍐垫槸鏈煡鐨�,骞朵笖鍙娴嬫鍦ㄨ�冩牳鐨勮澶囬伩鍏嶅浣欏伐鍗� List<TMonitorResult> monitorList = monitorMapper.getDistinctIP(); + //琛ュ厖閿欒鏃堕棿鐐� + Query onlineQuery = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); + Map<String, TMonitorResult> mongoMap = mongoTemplate.find(onlineQuery, TMonitorResult.class) + .stream() + .collect(Collectors.toMap(TMonitorResult::getNo, Function.identity(), (existing, replacement) -> replacement)); + for (TMonitorResult result : monitorList) { + TMonitorResult mongoData = mongoMap.get(result.getNo()); + if(mongoData!=null){ + result.setOffLineTimeStr(mongoData.getOffLineTimeStr()); + } + } List<TMonitorResult> dataList = new ArrayList<>(48); Integer time = times; List<CompletableFuture<TMonitorResult>> futureList = monitorList.stream() @@ -195,34 +214,15 @@ OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, time); return thread.call(); // 鍋囪 OnlineCheckThread 瀹炵幇浜� Callable 鎺ュ彛 }, executorService) - .orTimeout(60, TimeUnit.SECONDS) - //鍑虹幇寮傚父 + .orTimeout(180, TimeUnit.SECONDS) .exceptionally(ex -> { if (ex instanceof TimeoutException) { log.error("浠诲姟鎵ц瓒呮椂:"+monitor.getIp()); } else { - log.error("浠诲姟鎵ц寮傚父:"+monitor.getIp() + ex); + log.error("浠诲姟鎵ц寮傚父:"+monitor.getIp() +ex); + ex.printStackTrace(); } - int checkTimes = 1; - int offLineTimes = 1; - Map<String, Object> map = (Map<String, Object>) redisTemplate.opsForHash().get(RedisConstant.ONLINE_KEY, monitor.getIp()); - if (!CollectionUtils.isEmpty(map)) { - checkTimes = (Integer) map.get("checkTimes") + 1; - offLineTimes = (Integer) map.get("offLineTimes"); - } - monitor.setPingOnline(Boolean.FALSE); - monitor.setCheckCount(checkTimes); - monitor.setOffLineCount(offLineTimes); - if (monitor.getOffLineCount() >= time) { - WorkOrder workOrder = new WorkOrder(); - workOrder.setSerialNumber(monitor.getNo()); - List<String> errList = new ArrayList<>(); - errList.add(ErrorType.DEVICE_OFFLINE.getValue()); - workOrder.setErrorTypeList(errList); - workOrder.setStatus(WorkOrderStatusEnum.DISTRIBUTED); - monitor.setWorkOrder(workOrder); - } - return monitor; // 杩斿洖澶辫触鐨勭粨鏋� + return null; })) .collect(Collectors.toList()); @@ -233,7 +233,7 @@ try { allOf.get(60, TimeUnit.SECONDS); // 缁欎簣棰濆鐨�5绉掓潵鏀堕泦缁撴灉 } catch (TimeoutException e) { - log.warn("閮ㄥ垎浠诲姟鏈湪鎸囧畾鏃堕棿鍐呭畬鎴�"); + log.error("閮ㄥ垎浠诲姟鏈湪鎸囧畾鏃堕棿鍐呭畬鎴�"); } catch (Exception e2){ log.error("鏁版嵁鏀堕泦寮傚父"+e2); } @@ -242,27 +242,67 @@ .map(CompletableFuture::join) .filter(Objects::nonNull) .collect(Collectors.toList()); - //绛涢�夊嚭ping绂荤嚎鐨勮澶囷紝鏇存敼鏁版嵁搴撲负绂荤嚎 + Date now = new Date(); - List<UpdateOnlineVO> willUpdateList = dataList.stream().filter(result->!result.getPingOnline()).map(item -> { - UpdateOnlineVO vo = new UpdateOnlineVO(); - vo.setOnline(ApiConstants.UY_OnlineSite_Offline); - vo.setIp(item.getIp()); - vo.setUpdateTime(now); - return vo; - }).collect(Collectors.toList()); - monitorMapper.updateOnline(willUpdateList); + List<String> offLineList = new ArrayList<>(); + List<String> onLineList = new ArrayList<>(); + //鏌ュ嚭鏁版嵁搴撶函杞﹁締鎴栫函浜鸿劯璁惧 +// List<String> serialNumbers = monitorMapper.selectCarOrFace().stream().map(TMonitor::getSerialNumber).collect(Collectors.toList()); + dataList.forEach(item->{ + if(item.getPingOnline()) { + onLineList.add(item.getIp()); + } else if(!item.getPingOnline()) { + //绛涢�夊嚭ping绂荤嚎鐨勮澶囷紝鏇存敼鏁版嵁搴撲负绂荤嚎 + offLineList.add(item.getIp()); + } + }); + if(!CollectionUtils.isEmpty(offLineList)) { + monitorMapper.batchUpdateOnline(offLineList, now, ApiConstants.UY_OnlineSite_Offline); + } + if(!CollectionUtils.isEmpty(onLineList)) { + monitorMapper.batchUpdateOnline(onLineList, now, ApiConstants.UY_OnlineSite_Online); + } //瀛樻斁鍒癿ongo if (!CollectionUtils.isEmpty(dataList)) { + List<TMonitorResult> mongoList = new ArrayList<>(); + dataList.forEach(item->{ + String monitorType = item.getMonitorType(); + //鍚屼竴涓澶囧涓被鍨嬫瘡涓被鍨嬪瓨涓�鏉℃暟鎹紝浠ユ鍦ㄤ繚鐣欐暟鎹殑鎯呭喌涓嬪尯鍒嗙渷鍘呮爣绛� + if(StringUtils.isNotEmpty(monitorType)){ + String[] monitors = monitorType.split("/"); + for (String type : monitors) { + TMonitorResult mongoData = new TMonitorResult(); + BeanUtils.copyProperties(item,mongoData); + mongoData.setMonitorType(type); + if("1".equals(type)){ + mongoData.setProvinceTag(mongoData.getProvinceTagVideo()); + }else if("2".equals(type)){ + mongoData.setProvinceTag(mongoData.getProvinceTagCar()); + }else if("3".equals(type)){ + mongoData.setProvinceTag(mongoData.getProvinceTagFace()); + } + mongoList.add(mongoData); + } + } + }); //濡傛灉瀛樺湪涔嬪墠鐨勬暟鎹厛鍒犻櫎 Query query = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, TMonitorResult.class); //瀛樻斁鍦╩ongo涓� - mongoTemplate.insertAll(dataList); + mongoTemplate.insertAll(mongoList); } - //宸ュ崟 + //宸ュ崟锛堝悓涓�IP鍙敓鎴愪竴涓伐鍗曪級 + //鏌ヨ鏁版嵁搴撳凡瀛樺湪鐨勭绾垮伐鍗曡幏鍙杋p闆嗗悎锛屽墧闄� + List<String> ips = workOrderMapper.getOfflineWorkOrder(); List<WorkOrder> workOrderList = dataList.stream() - .filter(item -> Objects.nonNull(item.getWorkOrder())) + .filter(item -> Objects.nonNull(item.getWorkOrder()) && (CollectionUtils.isEmpty(ips) || !ips.contains(item.getIp()))) + .collect(Collectors.toMap( + TMonitorResult::getIp, + Function.identity(), + (existing, replacement) -> existing // 濡傛灉閬囧埌鐩稿悓鐨� IP锛屼繚鐣欑涓�涓� TMonitorResult 瀵硅薄 + )) + .values() + .stream() .map(TMonitorResult::getWorkOrder) .collect(Collectors.toList()); if (!CollectionUtils.isEmpty(workOrderList)) { @@ -299,7 +339,7 @@ item.setNo(item.getDeviceId()); } }); - pointService.setDeviceTagByGB(records); + pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Video); //瀛樻斁鍦╩ongo涓� mongoTemplate.insertAll(records); //鏇存柊point琛ㄥ湪绾跨姸鎬� @@ -311,9 +351,15 @@ vo.setUpdateTime(now); return vo; }).collect(Collectors.toList()); - monitorMapper.updateOnline(willUpdateList); - //绂荤嚎鐢熸垚宸ュ崟 - List<VideoOnlineResult> workOrders = records.stream().filter(item -> ApiConstants.UY_OnlineSite_Offline.equals(item.getStatus())).collect(Collectors.toList()); + monitorMapper.updateOnlineFromUyOrHk(willUpdateList); + //绂荤嚎鐢熸垚宸ュ崟,涓�涓猧p鍙敓鎴愪竴涓伐鍗� + List<VideoOnlineResult> workOrders = new ArrayList<>(records.stream() + .filter(item -> ApiConstants.UY_OnlineSite_Offline.equals(item.getStatus())) + .collect(Collectors.toMap( + VideoOnlineResult::getIpAddr, + Function.identity(), + (existing, replacement) -> existing // 濡傛灉閬囧埌鐩稿悓鐨� IP锛屼繚鐣欑涓�涓紙existing锛� + )).values()); uyErrorTypeCheckService.videoOnlineCheck(workOrders); } else { log.error("鐐逛綅鍦ㄧ嚎缁撴灉鏁版嵁涓虹┖{}", data); @@ -360,7 +406,7 @@ } }); //鎵撴爣绛� - pointService.setDeviceTagByGB(records); + pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Video); //瀛樻斁鍦╩ongo涓� mongoTemplate.insertAll(records); // // 宸ュ崟鐢熸垚 -- Gitblit v1.8.0