场景：
- 自研ETL项目同步每一个业务线不同微服务数据，需要设计一个任务正点检测数据源和同步数据是否一致。
- 每个同步实例有多张表，顺序执行时间很长。
- 因此，该场景需要使用多线程并行执行check任务。
/*** 初始化线程池*/
if( executorService == null ){executorService = new ThreadPoolExecutor(10, 10,60L, TimeUnit.SECONDS,new ArrayBlockingQueue(30));
}for (Map dbInstanceMap: dbInstanceList) {Future<List<SyncQualityResult>> result = executorService.submit(new Callable<List<SyncQualityResult>>() {@Overridepublic List<SyncQualityResult> call() {List<SyncQualityResult> results = new ArrayList<>();try {Long instanceId = Long.parseLong(dbInstanceMap.get("id").toString());logger.info(" -------- Start Quality Check With Instance :{} -------- " , instanceId);SyncInfo syncInfo = new SyncInfo();syncInfo.setInstanceId(instanceId);syncInfo.setServiceId(serviceId);PageData<SyncInfo> syncInfoPageData = syncInfoService.queryPage(serviceId, syncInfo, 1);// 查询每个实例的同步任务results = checkEverySyncTask(serviceId,dbInstanceMap,syncInfoPageData);logger.info(" -------- End Quality Check With Instance :{} -------- " , instanceId);}catch (InterruptedException ex){logger.error(ex.getMessage(),ex);return results;}catch (Exception ex) {ex.printStackTrace();}finally {return results;}}});futures.add(result);
}for (Future f : futures) {try {results.addAll((List<SyncQualityResult>) f.get(10,TimeUnit.SECONDS));} catch (InterruptedException | ExecutionException | TimeoutException ex) {ex.printStackTrace();}
}