countdownlatch

 public AjaxResult updateRsbtStationByMfId(RmbtFacilityEquipment rmbtFacilityEquipment) {String mfid = rmbtFacilityEquipment.getMfid();// 获取固定站信息RmbtFixedStation rmbtFixedStation = new RmbtFixedStation();rmbtFixedStation.setMfid(mfid);List<RmbtFixedStation> rmbtFixedStations = this.selectRmbtFixedStationAllList(rmbtFixedStation);rmbtFixedStation = rmbtFixedStations.get(0);rmbtFacilityEquipment.setRmbtFixedStation(rmbtFixedStation);// 分页参数优化int pageSize = 50; // 每页数据量int batchSize = 100; // 批量插入大小int threadCount = 6; // 线程数量long totalCount = rsbtStStationService.countRsbtStStation();int totalPages = (int) Math.ceil((double) totalCount / pageSize);// 创建线程安全的结果集合List<List<RsbtEquipmentRelation>> threadResults = new ArrayList<>();for (int i = 0; i < threadCount; i++) {threadResults.add(Collections.synchronizedList(new ArrayList<>()));}try {// 创建线程池ExecutorService executorService = Executors.newFixedThreadPool(threadCount);CountDownLatch latch = new CountDownLatch(totalPages);AtomicInteger processedCount = new AtomicInteger(0);// 分配页面给不同线程处理for (int pageNum = 0; pageNum < totalPages; pageNum++) {final int currentPage = pageNum;final int threadIndex = pageNum % threadCount;executorService.submit(() -> {try {// 分页查询数据List<RsbtStStation> pageStations = rsbtStStationService.selectRsbtStStationByPage(currentPage, pageSize);// 预先查询所有台站的频率信息Set<String> stationGuids = pageStations.stream().map(RsbtStStation::getGuid).collect(Collectors.toSet());// 批量查询频率信息List<RsbtStFreq> allFreqs = rsbtStFreqService.selectRsbtStFreqListByStationIds(new ArrayList<>(stationGuids));Map<String, List<RsbtStFreq>> freqMap = allFreqs.stream().collect(Collectors.groupingBy(RsbtStFreq::getStationGuid));// 处理每页数据List<RsbtEquipmentRelation> pageResults = new ArrayList<>();for (RsbtStStation rsbtStStation : pageStations) {List<RsbtStFreq> stationFreqs = freqMap.get(rsbtStStation.getGuid());if (stationFreqs != null && !stationFreqs.isEmpty()) {RsbtEquipmentRelation relation = createEquipmentRelation(rsbtStStation,stationFreqs.get(0),mfid,rmbtFacilityEquipment.getRmbtFixedStation());if (relation != null) {pageResults.add(relation);processedCount.incrementAndGet();}}}// 将结果添加到对应线程的结果集合threadResults.get(threadIndex).addAll(pageResults);// 打印处理进度log.info("已处理: {}/{}条数据, 当前页: {}/{}", processedCount.get(), totalCount, currentPage + 1, totalPages);} finally {latch.countDown();}});}// 等待所有线程完成latch.await();executorService.shutdown();// 合并所有线程的结果并批量插入for (List<RsbtEquipmentRelation> threadResult : threadResults) {for (int i = 0; i < threadResult.size(); i += batchSize) {int end = Math.min(i + batchSize, threadResult.size());List<RsbtEquipmentRelation> batch = threadResult.subList(i, end);rsbtEquipmentRelationService.batchInsertRsbtEquipmentRelation(batch);}}return AjaxResult.success("处理完成，共处理" + processedCount.get() + "条数据");} catch (Exception e) {log.error("数据处理异常", e);return AjaxResult.error("处理失败：" + e.getMessage());}}