Middleware/cn-data-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

将入库改成多线程

Browse Source
This commit is contained in:
hzj
2026-06-09 16:28:04 +08:00
parent 761fbc73f2
commit ff0cbc7151
1 changed files with 41 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/LnDataDealServiceImpl.java
View File

@@ -16,10 +16,14 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Comparator; import java.util.Comparator;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@@ -66,22 +70,45 @@ public class LnDataDealServiceImpl implements LnDataDealService {
private LineFeignClient lineFeignClient; private LineFeignClient lineFeignClient;
@QueryBean @QueryBean
private IPqsCommunicate iPqsCommunicate; private IPqsCommunicate iPqsCommunicate;
@Resource(name="asyncExecutor")
private Executor executor;
@Override @Override
public void batchInsertion(LnDataDTO lnDataDTO) { public void batchInsertion(LnDataDTO lnDataDTO) {
dataVService.batchInsertion(lnDataDTO.getDataVList());
dataHarmrateVQuery.batchInsertion(lnDataDTO.getDataHarmrateVDTOList()); List<CompletableFuture<Void>> futures = new ArrayList<>();
dataFlickerQuery.batchInsertion(lnDataDTO.getDataFlickerDTOList()); // 提交每个批量插入任务
dataFlucQuery.batchInsertion(lnDataDTO.getDataFlucDTOList()); futures.add(CompletableFuture.runAsync(() -> dataVService.batchInsertion(lnDataDTO.getDataVList()), executor));
dataHarmphasicIQuery.batchInsertion(lnDataDTO.getDataHarmphasicIDTOList()); futures.add(CompletableFuture.runAsync(() -> dataHarmrateVQuery.batchInsertion(lnDataDTO.getDataHarmrateVDTOList()), executor));
dataHarmphasicVQuery.batchInsertion(lnDataDTO.getDataHarmphasicVDTOList()); futures.add(CompletableFuture.runAsync(() -> dataFlickerQuery.batchInsertion(lnDataDTO.getDataFlickerDTOList()), executor));
dataHarmpowerPService.batchInsertion(lnDataDTO.getDataHarmpowerPDTOList()); futures.add(CompletableFuture.runAsync(() -> dataFlucQuery.batchInsertion(lnDataDTO.getDataFlucDTOList()), executor));
dataHarmpowerQService.batchInsertion(lnDataDTO.getDataHarmpowerQDTOList()); futures.add(CompletableFuture.runAsync(() -> dataHarmphasicIQuery.batchInsertion(lnDataDTO.getDataHarmphasicIDTOList()), executor));
dataHarmpowerSService.batchInsertion(lnDataDTO.getDataHarmpowerSDTOList()); futures.add(CompletableFuture.runAsync(() -> dataHarmphasicVQuery.batchInsertion(lnDataDTO.getDataHarmphasicVDTOList()), executor));
dataIService.batchInsertion(lnDataDTO.getDataIDTOList()); futures.add(CompletableFuture.runAsync(() -> dataHarmpowerPService.batchInsertion(lnDataDTO.getDataHarmpowerPDTOList()), executor));
dataInharmIService.batchInsertion(lnDataDTO.getDataInharmIDTOList()); futures.add(CompletableFuture.runAsync(() -> dataHarmpowerQService.batchInsertion(lnDataDTO.getDataHarmpowerQDTOList()), executor));
dataInharmVService.batchInsertion(lnDataDTO.getDataInharmVDTOList()); futures.add(CompletableFuture.runAsync(() -> dataHarmpowerSService.batchInsertion(lnDataDTO.getDataHarmpowerSDTOList()), executor));
dataPltService.batchInsertion(lnDataDTO.getDataPltDTOList()); futures.add(CompletableFuture.runAsync(() -> dataIService.batchInsertion(lnDataDTO.getDataIDTOList()), executor));
futures.add(CompletableFuture.runAsync(() -> dataInharmIService.batchInsertion(lnDataDTO.getDataInharmIDTOList()), executor));
futures.add(CompletableFuture.runAsync(() -> dataInharmVService.batchInsertion(lnDataDTO.getDataInharmVDTOList()), executor));
futures.add(CompletableFuture.runAsync(() -> dataPltService.batchInsertion(lnDataDTO.getDataPltDTOList()), executor));
// 等待所有任务完成，如果任一任务抛出异常，这里会重新抛出
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
// 关闭线程池（优雅关闭，等待已提交任务完成）
// dataVService.batchInsertion(lnDataDTO.getDataVList());
// dataHarmrateVQuery.batchInsertion(lnDataDTO.getDataHarmrateVDTOList());
// dataFlickerQuery.batchInsertion(lnDataDTO.getDataFlickerDTOList());
// dataFlucQuery.batchInsertion(lnDataDTO.getDataFlucDTOList());
// dataHarmphasicIQuery.batchInsertion(lnDataDTO.getDataHarmphasicIDTOList());
// dataHarmphasicVQuery.batchInsertion(lnDataDTO.getDataHarmphasicVDTOList());
// dataHarmpowerPService.batchInsertion(lnDataDTO.getDataHarmpowerPDTOList());
// dataHarmpowerQService.batchInsertion(lnDataDTO.getDataHarmpowerQDTOList());
// dataHarmpowerSService.batchInsertion(lnDataDTO.getDataHarmpowerSDTOList());
// dataIService.batchInsertion(lnDataDTO.getDataIDTOList());
// dataInharmIService.batchInsertion(lnDataDTO.getDataInharmIDTOList());
// dataInharmVService.batchInsertion(lnDataDTO.getDataInharmVDTOList());
// dataPltService.batchInsertion(lnDataDTO.getDataPltDTOList());
//更新mysqldevice表最新数据时间 //更新mysqldevice表最新数据时间
if(!CollectionUtils.isEmpty(lnDataDTO.getDataVList())){ if(!CollectionUtils.isEmpty(lnDataDTO.getDataVList())){