Index: ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/service/PfNotionalPoolingCoreService.java =================================================================== --- ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/service/PfNotionalPoolingCoreService.java (revision 34158) +++ ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/service/PfNotionalPoolingCoreService.java (revision 34159) @@ -12,7 +12,11 @@ import java.util.Map; import java.util.TimeZone; import java.util.UUID; +import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import javax.annotation.Resource; @@ -93,6 +97,7 @@ import com.freshport.freight.settlement.utils.SettlementInfo; import com.freshport.freight.settlement.utils.pffactory.AnalysisSettlementDataFactory; import com.freshport.freight.settlement.utils.pffactory.base.AnalysisJsonToEntity; +import com.freshport.freight.settlement.utils.threadpool.SettlementMainCallable; import com.github.pagehelper.PageHelper; import com.github.pagehelper.PageInfo; @@ -230,6 +235,9 @@ @Resource private JnOperateLogMapper jnOperateLogMapper; + @Resource(name = "threadPoolExecutor") + private ThreadPoolExecutor threadPoolExecutor; + /** * transJsonToVo(传入字符串转为对应对象) * @param jsonStr @@ -3050,7 +3058,7 @@ npIds.add(vo.getNpId()); } - CountDownLatch count = new CountDownLatch(10); + // CountDownLatch count = new CountDownLatch(10); Map queryParam = new HashMap(); queryParam.put("ywbhList", ywbhList); queryParam.put("npIds", npIds); @@ -3057,15 +3065,39 @@ queryParam.put("nationalList", list); queryParam.put("dataSource", param.getDataSource()); queryParam.put("exportFlag", CommonInfo.COMMON_YES); - MyThread[] threadArray = new MyThread[10]; + // MyThread[] threadArray = new MyThread[10]; + // for (int i = 0; i < 10; i++) { + // threadArray[i] = new MyThread(notionalPoolingService, count, + // queryParam, list, i + ""); + // threadArray[i].start(); + // } + // try { + // count.await(); + // } catch (InterruptedException e) { + // e.printStackTrace(); + // } + // 使用线程池替换原有方式 + List> futureList = new ArrayList<>(); for (int i = 0; i < 10; i++) { - threadArray[i] = new MyThread(notionalPoolingService, count, queryParam, list, i + ""); - threadArray[i].start(); + try { + Callable myCallable = new SettlementMainCallable(notionalPoolingService, list, queryParam, + String.valueOf(i)); + // 提交任务到线程池 + Future future = threadPoolExecutor.submit(myCallable); + // 将返回值 future 添加到 list，我们可以通过 future 获得 执行 Callable 得到的返回值 + futureList.add(future); + } catch (Exception e) { + e.printStackTrace(); + } } - try { - count.await(); - } catch (InterruptedException e) { - e.printStackTrace(); + // 等待所有线程返回 + for (Future fut : futureList) { + try { + // 获取任务时，添加超时时间， 如果超时则释放线程， 防止线程阻塞耗尽。 + System.out.println("-------> " + fut.get(60, TimeUnit.SECONDS)); + } catch (Exception e) { + e.printStackTrace(); + } } for (CostAccumulationVo vo : list) { @@ -3081,15 +3113,16 @@ && !PfSettlementInfo.PF_TRANS_YTPE_AIR.equals(vo.getTransType())) { vo.setReturnMarketTime(vo.getReturnMarketTime2()); } - + // 处理backColor - if (!CommonUtils.isEmpty(vo.getFeeType()) && (vo.getFeeType().contains("海运费")) || "合并结算".equals(vo.getFeeType())) { - if (vo.getBusinessNo().contains("TH")) { - vo.setBackColor(CommonInfo.COMMON_NO); - } - if (CommonInfo.COMMON_NO.equals(vo.getSfcjbj()) || CommonInfo.COMMON_NO.equals(vo.getYfcjbj())) { - vo.setBackColor(CommonInfo.COMMON_YES); - } + if (!CommonUtils.isEmpty(vo.getFeeType()) && (vo.getFeeType().contains("海运费")) + || "合并结算".equals(vo.getFeeType())) { + if (vo.getBusinessNo().contains("TH")) { + vo.setBackColor(CommonInfo.COMMON_NO); + } + if (CommonInfo.COMMON_NO.equals(vo.getSfcjbj()) || CommonInfo.COMMON_NO.equals(vo.getYfcjbj())) { + vo.setBackColor(CommonInfo.COMMON_YES); + } } } } Index: ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/utils/threadpool/SettlementMainCallable.java =================================================================== --- ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/utils/threadpool/SettlementMainCallable.java (nonexistent) +++ ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/utils/threadpool/SettlementMainCallable.java (revision 34159) @@ -0,0 +1,48 @@ +package com.freshport.freight.settlement.utils.threadpool; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; + +import javax.annotation.Resource; + +import com.freshport.freight.settlement.entity.vo.CostAccumulationVo; +import com.freshport.freight.settlement.service.NotionalPoolingService; + +/** + * 项目名称：ff-settlement + * 类名称：SettlementMainCallable + * 类描述：替代原来线程类 + * 创建人：wuwenjin + * 创建时间：2023年5月19日 下午4:01:16 + * 修改人：wuwenjin + * 修改时间：2023年5月19日 下午4:01:16 + * 修改备注： + * @version + * + */ +public class SettlementMainCallable implements Callable { + @Resource + private NotionalPoolingService service; + + private List list; + + private Map queryParam; + + private String i; + + public SettlementMainCallable(NotionalPoolingService service, List list, + Map queryParam, String i) { + this.service = service; + this.list = list; + this.queryParam = queryParam; + this.i = i; + } + + @Override + public String call() throws Exception { + service.queryNationalListInfo(queryParam, list, i); + return i; + } + +} Index: ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/utils/threadpool/SettlementThreadPoolExecutor.java =================================================================== --- ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/utils/threadpool/SettlementThreadPoolExecutor.java (nonexistent) +++ ff-cloud/ff-settlement/src/main/java/com/freshport/freight/settlement/utils/threadpool/SettlementThreadPoolExecutor.java (revision 34159) @@ -0,0 +1,42 @@ +package com.freshport.freight.settlement.utils.threadpool; + +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * 项目名称：ff-settlement + * 类名称：SettlementThreadPoolExecutor + * 类描述： 结算系统线程池配置 + * 创建人：wuwenjin + * 创建时间：2023年5月19日 上午9:50:08 + * 修改人：wuwenjin + * 修改时间：2023年5月19日 上午9:50:08 + * 修改备注： + * @version 1.0 + */ +@Configuration +public class SettlementThreadPoolExecutor { + // 核心线程数为 20 + private static final int CORE_POOL_SIZE = 20; + + // 最大线程数 100 + private static final int MAX_POOL_SIZE = 100; + + // 任务队列为 LinkedBlockingQueue，并且容量为 500; + private static final int QUEUE_CAPACITY = 500; + + // 等待时间为 30分钟。 + private static final Long KEEP_ALIVE_TIME = 1800L; + + // 丢弃任务并抛出RejectedExecutionException异常。(默认拒绝策略) + + @Bean(name = "threadPoolExecutor") + public ThreadPoolExecutor threadPoolExecutor() { + return new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, + new LinkedBlockingQueue<>(QUEUE_CAPACITY), new ThreadPoolExecutor.AbortPolicy()); + } +}