| | |
| | | } |
| | | |
| | | /** |
| | | * 点位在线检测 |
| | | * 点位在线PING检测 |
| | | */ |
| | | //TODO:点位在线(修改逻辑只针对工单) |
| | | public void pointOnline() throws ExecutionException, InterruptedException { |
| | | log.info("开始检测点位在线"); |
| | | Integer times = 2; |
| | |
| | | } else { |
| | | log.error("请配置离线次数,此次设置为默认值2"); |
| | | } |
| | | // 先查出设备IP集合 |
| | | // 先查出设备IP集合,剔除掉在线情况是未知的,并且只检测正在考核的设备避免多余工单 |
| | | List<TMonitorResult> monitorList = monitorMapper.getDistinctIP(); |
| | | // List<Future<TMonitorResult>> futureList = new ArrayList<>(48); |
| | | List<TMonitorResult> dataList = new ArrayList<>(48); |
| | | // for (TMonitorResult monitor : monitorList) { |
| | | // OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, times); |
| | | // Future<TMonitorResult> future = executorService.submit(thread); |
| | | // futureList.add(future); |
| | | // } |
| | | // for (Future<TMonitorResult> future : futureList) { |
| | | // dataList.add(future.get()); // get方法会阻塞,知道拿到结果才继续执行for |
| | | // } |
| | | Integer time = times; |
| | | List<CompletableFuture<TMonitorResult>> futureList = monitorList.stream() |
| | | .map(monitor -> CompletableFuture.supplyAsync(() -> { |
| | | OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, time); |
| | | return thread.call(); // 假设 OnlineCheckThread 实现了 Callable 接口 |
| | | }, executorService) |
| | | .orTimeout(30, TimeUnit.SECONDS) |
| | | .orTimeout(60, TimeUnit.SECONDS) |
| | | //出现异常 |
| | | .exceptionally(ex -> { |
| | | if (ex instanceof TimeoutException) { |
| | | log.error("任务执行超时:"); |
| | | log.error("任务执行超时:"+monitor.getIp()); |
| | | } else { |
| | | log.error("任务执行异常:" + ex); |
| | | log.error("任务执行异常:"+monitor.getIp() + ex); |
| | | } |
| | | int checkTimes = 1; |
| | | int offLineTimes = 1; |
| | | //TODO:BUG |
| | | 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.setOnline(Boolean.FALSE); |
| | | monitor.setPingOnline(Boolean.FALSE); |
| | | monitor.setCheckCount(checkTimes); |
| | | monitor.setOffLineCount(offLineTimes); |
| | | if (monitor.getOffLineCount() >= time) { |
| | |
| | | futureList.toArray(new CompletableFuture[0]) |
| | | ); |
| | | try { |
| | | allOf.get(35, TimeUnit.SECONDS); // 给予额外的5秒来收集结果 |
| | | allOf.get(60, TimeUnit.SECONDS); // 给予额外的5秒来收集结果 |
| | | } catch (TimeoutException e) { |
| | | log.warn("部分任务未在指定时间内完成"); |
| | | } catch (Exception e2){ |
| | |
| | | |
| | | dataList = futureList.stream() |
| | | .map(CompletableFuture::join) |
| | | .filter(result -> result != null) |
| | | .filter(Objects::nonNull) |
| | | .collect(Collectors.toList()); |
| | | // 更新point表的在线标识 |
| | | //筛选出ping离线的设备,更改数据库为离线 |
| | | Date now = new Date(); |
| | | List<UpdateOnlineVO> willUpdateList = dataList.stream().map(item -> { |
| | | List<UpdateOnlineVO> willUpdateList = dataList.stream().filter(result->!result.getPingOnline()).map(item -> { |
| | | UpdateOnlineVO vo = new UpdateOnlineVO(); |
| | | vo.setOnline(item.getOnline()); |
| | | vo.setOnline(ApiConstants.UY_OnlineSite_Offline); |
| | | vo.setIp(item.getIp()); |
| | | vo.setUpdateTime(now); |
| | | return vo; |
| | | }).collect(Collectors.toList()); |
| | | monitorMapper.updateOnline(willUpdateList); |
| | | // 工单 |
| | | List<WorkOrder> workOrderList = dataList.stream() |
| | | .filter(item -> Objects.nonNull(item.getWorkOrder())) |
| | | .map(TMonitorResult::getWorkOrder) |
| | | .collect(Collectors.toList()); |
| | | if (!CollectionUtils.isEmpty(workOrderList)) { |
| | | workOrderService.innerAddWorkOrder(workOrderList); |
| | | } |
| | | //存放到mongo |
| | | if (!CollectionUtils.isEmpty(dataList)) { |
| | | //如果存在之前的数据先删除 |
| | |
| | | //存放在mongo中 |
| | | mongoTemplate.insertAll(dataList); |
| | | } |
| | | //工单 |
| | | List<WorkOrder> workOrderList = dataList.stream() |
| | | .filter(item -> Objects.nonNull(item.getWorkOrder())) |
| | | .map(TMonitorResult::getWorkOrder) |
| | | .collect(Collectors.toList()); |
| | | if (!CollectionUtils.isEmpty(workOrderList)) { |
| | | workOrderService.innerAddWorkOrder(workOrderList); |
| | | } |
| | | log.info("点位在线监测完成"); |
| | | } |
| | | |
| | | //(弃用)点位在线率 |
| | | //点位在线率(优云) |
| | | public void videoOnlineTask() { |
| | | //视频图像质量 |
| | | log.info("开始执行点位在线数据同步"); |
| | |
| | | Query query = new Query(Criteria |
| | | .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); |
| | | DeleteResult result = mongoTemplate.remove(query, VideoOnlineResult.class); |
| | | //打标签 |
| | | records.forEach(item -> { |
| | | if (Objects.nonNull(item.getDeviceId())) { |
| | | item.setNo(item.getDeviceId()); |
| | | } |
| | | }); |
| | | pointService.setDeviceTagByGB(records); |
| | | //存放在mongo中 |
| | | mongoTemplate.insertAll(records); |
| | | // 工单生成 |
| | | uyErrorTypeCheckService.videoOnlineCheck(records); |
| | | //更新point表在线状态 |
| | | Date now = new Date(); |
| | | List<UpdateOnlineVO> willUpdateList = records.stream().map(item -> { |
| | | UpdateOnlineVO vo = new UpdateOnlineVO(); |
| | | vo.setOnline(item.getStatus()); |
| | | vo.setIp(item.getIpAddr()); |
| | | 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()); |
| | | uyErrorTypeCheckService.videoOnlineCheck(workOrders); |
| | | } else { |
| | | log.error("点位在线结果数据为空{}", data); |
| | | } |
| | |
| | | item.setNo(item.getDeviceId()); |
| | | } |
| | | }); |
| | | //打标签 |
| | | pointService.setDeviceTagByGB(records); |
| | | //存放在mongo中 |
| | | mongoTemplate.insertAll(records); |
| | | // 工单生成 |
| | | uyErrorTypeCheckService.recordMetaDSumCheck(records); |
| | | // // 工单生成 |
| | | // uyErrorTypeCheckService.recordMetaDSumCheck(records); |
| | | } |
| | | } else { |
| | | log.error("录像可用数据为空{}", jsonObject); |