这里是我的主线程,在每个进程中,我都在调用执行服务线程,在第4行,在更新状态之前,我想等待我的所有线程完成。在第4行,在更新状态之前,我想等待所有线程完成。
line 1- missingStrategyContext.process(FileType.CARD, programId, fromDate, toDate, channel);
line 2- missingStrategyContext.process(FileType.TRANSACTION, programId, fromDate, toDate, channel);
line 3- missingStrategyContext.process(FileType.REFUND, programId, fromDate, toDate, channel);
line 4- processingLog.setProcessingStatus(TransactionState.SUCCESS.name());
在每一个过程中,我都在决定我必须调用的文件类型值的策略。
public void process(FileType fileType, String programId, Long fromDate, Long toDate, String channel){
map.get(fileType).process(programId, fromDate, toDate, channel, fileType);
}
然后根据文件类型,我实现了我的进程方法,并在每个文件类型的实现中调用执行器服务。
@Override
public void process(String programId, Long fromDate, Long toDate, String channel, FileType fileType) {
MultiTenantTxMgmt multiTenantTxMgmt = MultiTenantTxMgmtUtil.get(programId);
EntityManager entityManager = null;
List<MissingReason> reasonList = new ArrayList<>();
try {
entityManager = multiTenantTxMgmt.getEntityManager();
reasonList = missingDataRepository.getDistinctMissingReasonForFileType(entityManager, fileType, TransactionState.READYTOATTEMPT);
}catch (Exception exception) {
logger.error(exception.getMessage(), exception);
throw new UnkownBatchSaveException();
} finally {
entityManager.close();
}
reasonList.forEach(
missingReason -> deExecutorService.dotask(() ->
missingTransactionStrategyContext.processMissingV3(missingReason, programId, fromDate, toDate, channel, fileType)
)
);
}
你可以使用 CountDownLatch#await
. 例如从文档中复制。
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(N);
for (int i = 0; i < N; ++i) // create and start threads
new Thread(new Worker(startSignal, doneSignal)).start();
doSomethingElse(); // don't let run yet
startSignal.countDown(); // let all threads proceed
doSomethingElse();
doneSignal.await(); // wait for all to finish
你可以让 doTask
方法让返回一个(可完成的)未来(如果它还没有的话),并在方法中返回一个期货列表 process
例如,替换
reasonList.forEach(
missingReason -> deExecutorService.dotask(() ->
missingTransactionStrategyContext.processMissingV3(missingReason, programId, fromDate, toDate, channel, fileType)
)
);
与
final List<Future> futures = reasonList.stream()
.map(missingReason -> deExecutorService.dotask(() ->
missingTransactionStrategyContext
.processMissingV3(missingReason, programId, fromDate, toDate, channel, fileType)))
.collect(Collectors.toList());
}
return futures;
那么在你的调用部分,你可以等这些期货全部终止后再继续。