优化百度和face++代码

examcloud-core-oe-face-starter/src/main/java/cn/com/qmth/examcloud/core/oe/student/face/starter/config/ProcessBaiduFacelivenessTask.java

@@ -24,83 +24,89 @@ import cn.com.qmth.examcloud.web.bootstrap.PropertyHolder;
 @Order(201)
 public class ProcessBaiduFacelivenessTask implements ApplicationRunner {
 
-	private static ExamCloudLog captureLog = ExamCloudLogFactory
-			.getLog("PROCESS_EXAM_CAPTURE_TASK_LOGGER");
-
-	@Autowired
-	ExamCaptureQueueRepo examCaptureQueueRepo;
-
-	private void process(ConcurrentTask<ExamCaptureQueueEntity> concurrentTask,
-			String processBatchNum) {
-
-		// 如果队列没满，则从数据库中查数据并插入
-		List<ExamCaptureQueueEntity> examCaptureQueueList = examCaptureQueueRepo
-				.findNeedFacelivenessDetectExamCaptureQueuesLimit(
-						PropertyHolder.getInt("$capture.queue.limit", 100), processBatchNum);
-
-		if (null == examCaptureQueueList || examCaptureQueueList.isEmpty()) {
-			captureLog
-					.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum + "] 抓拍队列中没有取到数据，2秒后重试");
-			return;
-		}
-
-		captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum + "] 抓拍队列中的数据条数为："
-				+ examCaptureQueueList.size());
-
-		for (ExamCaptureQueueEntity offeredQueueEntity : examCaptureQueueList) {
-			while (true) {
-				boolean offerSuccess = concurrentTask.offerElement(offeredQueueEntity);
-				if (offerSuccess) {
-					offeredQueueEntity.setProcessBatchNum(processBatchNum);
-					examCaptureQueueRepo.save(offeredQueueEntity);
-
-					captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum
-							+ "] 向工作队列中添加数据成功：fileUrl=" + offeredQueueEntity.getFileUrl());
-					break;
-				}
-
-				captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum
-						+ "] 向工作队列中添加数据失败，30秒后重试：fileUrl=" + offeredQueueEntity.getFileUrl());
-
-				Util.sleep(PropertyHolder.getInt("$capture.queue.offer.sleepSeconds.", 30));
-			}
-		}
-
-	}
-
-	@Override
-	public void run(ApplicationArguments args) {
-
-		ConcurrentTask<ExamCaptureQueueEntity> concurrentTask = new ConcurrentTask<ExamCaptureQueueEntity>();
-		concurrentTask.setMaxActiveThreadSize(
-				PropertyHolder.getInt("$capture.thread.maxActiveThreadSize", 100));
-		concurrentTask.setMinThreadSize(PropertyHolder.getInt("$capture.thread.minThreadSize", 2));
-		concurrentTask.setWorker(new BaiduFaceLivenessWorker());
-		concurrentTask.start();
-		// 当前获取数据的批次号(默认用时间戳)
-		String processBatchNum = "B_" + System.currentTimeMillis();
-
-		captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum + "] 启动百度处理服务...");
-
-		Thread thread = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				while (true) {
-					try {
-						process(concurrentTask, processBatchNum);
-						Util.sleep(PropertyHolder.getInt("$capture.queue.read.sleepSeconds.", 2));
-					} catch (Exception e) {
-						captureLog.error("[PROCESS_FACEPP." + processBatchNum + "] 百度活体检测出异常,3秒后重试",
-								e);
-						Util.sleep(3);
-					}
-				}
-
-			}
-		});
-
-		thread.setDaemon(true);
-		thread.start();
-	}
+    private static ExamCloudLog captureLog = ExamCloudLogFactory
+            .getLog("PROCESS_EXAM_CAPTURE_TASK_LOGGER");
+
+    @Autowired
+    ExamCaptureQueueRepo examCaptureQueueRepo;
+
+    private int process(ConcurrentTask<ExamCaptureQueueEntity> concurrentTask,
+                        String processBatchNum, Integer limit) {
+
+        // 如果队列没满，则从数据库中查数据并插入
+        List<ExamCaptureQueueEntity> examCaptureQueueList = examCaptureQueueRepo
+                .findNeedFacelivenessDetectExamCaptureQueuesLimit(limit, processBatchNum);
+
+        if (null == examCaptureQueueList || examCaptureQueueList.isEmpty()) {
+            captureLog
+                    .debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum + "] 抓拍队列中没有取到数据，2秒后重试");
+            return 0;
+        }
+
+        captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum + "] 抓拍队列中的数据条数为："
+                + examCaptureQueueList.size());
+
+        for (ExamCaptureQueueEntity offeredQueueEntity : examCaptureQueueList) {
+            while (true) {
+                boolean offerSuccess = concurrentTask.offerElement(offeredQueueEntity);
+                if (offerSuccess) {
+                    offeredQueueEntity.setProcessBatchNum(processBatchNum);
+                    examCaptureQueueRepo.save(offeredQueueEntity);
+
+                    captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum
+                            + "] 向工作队列中添加数据成功：fileUrl=" + offeredQueueEntity.getFileUrl());
+                    break;
+                }
+
+                captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum
+                        + "] 向工作队列中添加数据失败，30秒后重试：fileUrl=" + offeredQueueEntity.getFileUrl());
+
+                Util.sleep(PropertyHolder.getInt("$capture.queue.offer.sleepSeconds.", 30));
+            }
+        }
+        return examCaptureQueueList.size();
+    }
+
+    @Override
+    public void run(ApplicationArguments args) {
+
+        ConcurrentTask<ExamCaptureQueueEntity> concurrentTask = new ConcurrentTask<ExamCaptureQueueEntity>();
+        concurrentTask.setMaxActiveThreadSize(
+                PropertyHolder.getInt("$capture.thread.maxActiveThreadSize", 100));
+        concurrentTask.setMinThreadSize(PropertyHolder.getInt("$capture.thread.minThreadSize", 2));
+        concurrentTask.setWorker(new BaiduFaceLivenessWorker());
+        concurrentTask.setMaxActiveThreadSize(PropertyHolder.getInt("$capture.baidu.thread.maxActiveThreadSize", 30));
+        concurrentTask.start();
+        // 当前获取数据的批次号(默认用时间戳)
+        String processBatchNum = "B_" + System.currentTimeMillis();
+
+        captureLog.debug("[PROCESS_BAIDUFACELIVENESS." + processBatchNum + "] 启动百度处理服务...");
+
+        Thread thread = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                while (true) {
+                    try {
+                        Integer limit = PropertyHolder.getInt("$capture.queue.limit", 100);
+                        for (int i = 0; i < 5; i++) {
+                            int realCount = process(concurrentTask, processBatchNum, limit);
+                            if (realCount < limit) {
+                                break;
+                            }
+                        }
+                        Util.sleep(PropertyHolder.getInt("$capture.queue.read.sleepSeconds.", 2));
+                    } catch (Exception e) {
+                        captureLog.error("[PROCESS_FACEPP." + processBatchNum + "] 百度活体检测出异常,3秒后重试",
+                                e);
+                        Util.sleep(3);
+                    }
+                }
+
+            }
+        });
+
+        thread.setDaemon(true);
+        thread.start();
+    }
 
 }

examcloud-core-oe-face-starter/src/main/java/cn/com/qmth/examcloud/core/oe/student/face/starter/config/ProcessFaceCompareQueueTask.java

@@ -25,83 +25,89 @@ import cn.com.qmth.examcloud.web.bootstrap.PropertyHolder;
 @Order(200)
 public class ProcessFaceCompareQueueTask implements ApplicationRunner {
 
-	private final ExamCloudLog captureLog = ExamCloudLogFactory
-			.getLog("PROCESS_EXAM_CAPTURE_TASK_LOGGER");
-
-	@Autowired
-	ExamCaptureQueueRepo examCaptureQueueRepo;
-
-	@Autowired
-	SmsCloudService smsCloudService;
-
-	private void process(ConcurrentTask<ExamCaptureQueueEntity> concurrentTask,
-			String processBatchNum) {
-		// 如果队列没满，则从数据库中查数据并插入
-		List<ExamCaptureQueueEntity> examCaptureQueueList = examCaptureQueueRepo
-				.findNeedFaceCompareExamCaptureQueuesLimitByProcessBatchNum(
-						PropertyHolder.getInt("$capture.queue.limit", 100), processBatchNum);
-
-		if (null == examCaptureQueueList || examCaptureQueueList.isEmpty()) {
-			captureLog.debug("[PROCESS_FACEPP." + processBatchNum + "] 抓拍队列中没有取到数据，2秒后重试");
-			return;
-		}
-
-		captureLog.debug("[PROCESS_FACEPP." + processBatchNum + "] 抓拍队列中的数据条数为："
-				+ examCaptureQueueList.size());
-
-		for (ExamCaptureQueueEntity offeredQueueEntity : examCaptureQueueList) {
-			while (true) {
-				boolean offerSuccess = concurrentTask.offerElement(offeredQueueEntity);
-				// 如果向队列中添加数据成功，则更新标识
-				if (offerSuccess) {
-					offeredQueueEntity.setProcessBatchNum(processBatchNum);
-					examCaptureQueueRepo.save(offeredQueueEntity);
-
-					captureLog.debug("[PROCESS_FACEPP." + processBatchNum
-							+ "] 向工作队列中添加数据成功：fileUrl=" + offeredQueueEntity.getFileUrl());
-					break;
-				}
-
-				captureLog.debug("[PROCESS_FACEPP." + processBatchNum
-						+ "] 向工作队列中添加数据失败，30秒后重试：fileUrl=" + offeredQueueEntity.getFileUrl());
-
-				Util.sleep(PropertyHolder.getInt("$capture.queue.offer.sleepSeconds.", 30));
-			}
-		}
-	}
-
-	@Override
-	public void run(ApplicationArguments args) {
-
-		ConcurrentTask<ExamCaptureQueueEntity> concurrentTask = new ConcurrentTask<ExamCaptureQueueEntity>();
-		concurrentTask.setMaxActiveThreadSize(
-				PropertyHolder.getInt("$capture.thread.maxActiveThreadSize", 100));
-		concurrentTask.setMinThreadSize(PropertyHolder.getInt("$capture.thread.minThreadSize", 2));
-		concurrentTask.setWorker(new FacePPCompareWorker());
-		concurrentTask.start();
-		// 当前获取数据的批次号(默认用时间戳)
-		String processBatchNum = "A_" + System.currentTimeMillis();
-
-		captureLog.debug("[PROCESS_FACEPP." + processBatchNum + "] 启动face++人脸比对服务...");
-
-		Thread thread = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				while (true) {
-					try {
-						process(concurrentTask, processBatchNum);
-						Util.sleep(PropertyHolder.getInt("$capture.queue.read.sleepSeconds.", 2));
-					} catch (Exception e) {
-						captureLog.error("[PROCESS_FACEPP." + processBatchNum + "] 百度活体检测出异常,3秒后重试",
-								e);
-						Util.sleep(3);
-					}
-				}
-
-			}
-		});
-
-		thread.setDaemon(true);
-		thread.start();
-	}
+    private final ExamCloudLog captureLog = ExamCloudLogFactory
+            .getLog("PROCESS_EXAM_CAPTURE_TASK_LOGGER");
+
+    @Autowired
+    ExamCaptureQueueRepo examCaptureQueueRepo;
+
+    @Autowired
+    SmsCloudService smsCloudService;
+
+    private int process(ConcurrentTask<ExamCaptureQueueEntity> concurrentTask,
+                        String processBatchNum, Integer limit) {
+        // 如果队列没满，则从数据库中查数据并插入
+        List<ExamCaptureQueueEntity> examCaptureQueueList = examCaptureQueueRepo
+                .findNeedFaceCompareExamCaptureQueuesLimitByProcessBatchNum(limit, processBatchNum);
+
+        if (null == examCaptureQueueList || examCaptureQueueList.isEmpty()) {
+            captureLog.debug("[PROCESS_FACEPP." + processBatchNum + "] 抓拍队列中没有取到数据，2秒后重试");
+            return 0;
+        }
+
+        captureLog.debug("[PROCESS_FACEPP." + processBatchNum + "] 抓拍队列中的数据条数为："
+                + examCaptureQueueList.size());
+
+        for (ExamCaptureQueueEntity offeredQueueEntity : examCaptureQueueList) {
+            while (true) {
+                boolean offerSuccess = concurrentTask.offerElement(offeredQueueEntity);
+                // 如果向队列中添加数据成功，则更新标识
+                if (offerSuccess) {
+                    offeredQueueEntity.setProcessBatchNum(processBatchNum);
+                    examCaptureQueueRepo.save(offeredQueueEntity);
+
+                    captureLog.debug("[PROCESS_FACEPP." + processBatchNum
+                            + "] 向工作队列中添加数据成功：fileUrl=" + offeredQueueEntity.getFileUrl());
+                    break;
+                }
+
+                captureLog.debug("[PROCESS_FACEPP." + processBatchNum
+                        + "] 向工作队列中添加数据失败，30秒后重试：fileUrl=" + offeredQueueEntity.getFileUrl());
+
+                Util.sleep(PropertyHolder.getInt("$capture.queue.offer.sleepSeconds.", 30));
+            }
+        }
+        return examCaptureQueueList.size();
+    }
+
+    @Override
+    public void run(ApplicationArguments args) {
+
+        ConcurrentTask<ExamCaptureQueueEntity> concurrentTask = new ConcurrentTask<ExamCaptureQueueEntity>();
+        concurrentTask.setMaxActiveThreadSize(
+                PropertyHolder.getInt("$capture.thread.maxActiveThreadSize", 100));
+        concurrentTask.setWorker(new FacePPCompareWorker());
+        concurrentTask.setMaxActiveThreadSize(PropertyHolder.getInt("$capture.facePP.thread.maxActiveThreadSize", 30));
+        concurrentTask.start();
+        // 当前获取数据的批次号(默认用时间戳)
+        String processBatchNum = "A_" + System.currentTimeMillis();
+
+        captureLog.debug("[PROCESS_FACEPP." + processBatchNum + "] 启动face++人脸比对服务...");
+
+        Thread thread = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                while (true) {
+                    try {
+                        Integer limit = PropertyHolder.getInt("$capture.queue.limit", 100);
+                        for (int i = 0; i < 5; i++) {
+                            int realCount = process(concurrentTask, processBatchNum, limit);
+                            if (realCount < limit) {
+                                break;
+                            }
+                        }
+                        Util.sleep(PropertyHolder.getInt("$capture.queue.read.sleepSeconds.", 2));
+                    } catch (Exception e) {
+                        captureLog.error("[PROCESS_FACEPP." + processBatchNum + "] 百度活体检测出异常,3秒后重试",
+                                e);
+                        Util.sleep(3);
+                    }
+                }
+
+            }
+        });
+
+        thread.setDaemon(true);
+        thread.start();
+    }
 }