1、添加对于数据库无摄像头数据报错校验

src/main/java/com/censoft/flink/StreamingJob.java

@@ -80,6 +80,9 @@ public class StreamingJob {
             warnPushBase.setOrderName("AI预警平台");
             //获取摄像头名称
             AlgorithmCameraPo cameraPo = getAlgorithmCameraPoByRtsp(algorithmCameraPoList, algorithmPushDto.getCameraName());
+            if (cameraPo == null){
+                return null;
+            }
             warnPushBase.setAlarmDeviceName(cameraPo.getCameraName());
             //获取摄像头归属
             warnPushBase.setBelong(cameraPo.getBelong());
@@ -87,8 +90,12 @@ public class StreamingJob {
             warnPushBase.setChannelId(cameraPo.getRtsp()+"$1$0$0");
             warnPushBase.setGradeName("二级预警");
             //获取摄像头名称
-            warnPushBase.setGradeType(getAlgorithmPoById(algorithmPoList,Long.valueOf(algorithmPushDto.getAlgorithmName()))
-                    .getAlgorithmName());
+            String algorithmName = getAlgorithmPoById(algorithmPoList, Long.valueOf(algorithmPushDto.getAlgorithmName()))
+                    .getAlgorithmName();
+            warnPushBase.setGradeType(algorithmName);
+            if (cameraPo == null){
+                return null;
+            }
             warnPushBase.setAlarmTime(new Date(algorithmPushDto.getTimeStamp() * 1000));
             //获取图片地址
             warnPushBase.setPicture("http://"+algorithmPushDto.getServerIp()+algorithmPushDto.getPictureAddress());
@@ -99,6 +106,9 @@ public class StreamingJob {
             return JSON.toJSONString(warnPushBase);
         });
 
+        //去空
+        outputStreamOperator = outputStreamOperator.filter(algorithmPushString -> algorithmPushString!= null && "".equals(algorithmPushString));
+
         //3、5 输出kafka
         outputStreamOperator.addSink(new FlinkKafkaProducer("192.168.10.137:9092", "test-topic", new SimpleStringSchema()));
 
@@ -111,15 +121,25 @@ public class StreamingJob {
 
 
     private static AlgorithmCameraPo getAlgorithmCameraPoByRtsp(List<AlgorithmCameraPo> cameraPos, String rtsp) {
-        return cameraPos
+        Optional<AlgorithmCameraPo> first = cameraPos
                 .stream()
-                .filter(cameraPo -> rtsp.equals(cameraPo.getRtsp())).findFirst().get();
+                .filter(cameraPo -> rtsp.equals(cameraPo.getRtsp())).findFirst();
+        if (first.isPresent()) {
+            return first.get();
+        }else {
+            return null;
+        }
     }
 
 
     private static AlgorithmPo getAlgorithmPoById(List<AlgorithmPo> algorithmPos, Long id) {
-        return algorithmPos
+        Optional<AlgorithmPo> first = algorithmPos
                 .stream()
-                .filter(algorithmPo -> id.equals(algorithmPo.getId())).findFirst().get();
+                .filter(algorithmPo -> id.equals(algorithmPo.getId())).findFirst();
+        if (first.isPresent()) {
+            return first.get();
+        }else {
+            return null;
+        }
     }
 }