相关推荐recommended
java多线程带返回值的方式方法
作者:mmseoamin日期:2023-12-25

java利用线程池带有返回值的方式,大体逻辑批量处理大量数据,启用线程池,处理完成后将所有的返回内容进行组装拼接

废话不多说开始看代码,重点敲黑板:

1.ThreadPoolExecutor 线程池创建

2.CountDownLatch 同步工具类,让主线程一直等待,直到子线程执行完后再执行

java多线程带返回值的方式方法,在这里插入图片描述,第1张

3.listret 用于接收多线程返回值

方式一

使用线程

// 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(coresNumber * 2, coresNumber * 2 + 1, 1000, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
         /*创建List用来接收多线程返回的内容,泛型里面可以自定义,String或者对象亦或者其他类型*/
        List> listret = new ArrayList<>();
         // 同步工具类,让主线程一直等待,直到子线程执行完后再执行
        CountDownLatch downLatch = new CountDownLatch(partition.size());
        // 循环任务的List
        for (List stringList : partition) {
            // 启用开启多个线程
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 开始调用具体业务代码
                        Map mapRet = pmpTargetPriceService.targetPriceThreadTask(stringList, initiateTaskType, userName);
                        listret.add(mapRet);
                    } catch (Exception e) {
                        logger.error("循环开启线多线程报错,调用下游系统出现错误,异常:" + e);
                    } finally {
                        // 业务逻辑处理完毕,计数器减一【当前线程处理任务完毕,线程释放进入线程池,等待处理下一个任务】
                        downLatch.countDown();
                    }
                }
            });
        }
         // 主线程需要等待子任务线程执行完,结果汇总之后,主线程继续往下执行
        try {
            downLatch.await();
        } catch (Exception e) {
            logger.error("等待超时", e);
            throw new RuntimeException("系统处理超时,请稍后再试");
        }
        // 对返回组装的list进循环处理业务逻辑
        for (Map esbResultPlm1 : listret) {
            // 从Future对象上获取任务的返回值,并输出到控制台
//            Map esbResultPlm1 = (Map) f.get();
            // todo 对我返回的多个map进行拼接
            if (esbResultPlm1.get("status").equals("fail")) {
                failureNum = (int) esbResultPlm1.get("failureNum");
                failureMsg.append(esbResultPlm1.get("msg"));
                map.put("msg", failureMsg.toString());
                failureNumCount += failureNum;
            } else {
                successNum = (int) esbResultPlm1.get("successNum");
                successMsg.append(esbResultPlm1.get("msg"));
                map.put("msg", successMsg.toString());
                successNumCount += successNum;
            }
        }

方法一得到的结果如下,使用线程池我这里是核数乘以2是核心线程16,最大17,所以这里最多是16个线程,而且他是无序的随机分配的

java多线程带返回值的方式方法,在这里插入图片描述,第2张

java多线程带返回值的方式方法,在这里插入图片描述,第3张

方式二

重点不用线程池使用@Async注解,但是策略得有所调整,大体逻辑比如你待处理的数据有100条,你可以将这个List按10条为一个新的List,循环这个集合,在调用的实际方法上加@Async注解,从而实现多线程加快循环也是可以的

@Async注意点,加了该注解的方法不能再同一个类,否则无效,其次有可能存在启动过程@Async UnsatisfiedDependencyException导致 SpringBoot 无法启动问题解决,我这里是在报错的类里有注入service或者mapper的注解上加了@Lazy注解就可以

java多线程带返回值的方式方法,在这里插入图片描述,第4张

        // 将要发送的集合按10个一组进行重新组装
        List> partition = Lists.partition(list, 10);
         /*创建List用来接收多线程返回的内容,泛型里面可以自定义,String或者对象亦或者其他类型*/
        List> listret = new ArrayList<>();
        // 循环任务的List
        for (List stringList : partition) {
            // 开始调用具体业务代码
            Map mapRet = pmpTargetPriceService.targetPriceThreadTask(stringList, initiateTaskType, userName);
            listret.add(mapRet);
         }
        // 对返回组装的list进循环处理业务逻辑
        for (Map esbResultPlm1 : listret) {
           //对返回的内容进行业务处理
        }
        
// 调用的方法,返回map
 	@Async
    public Map targetPriceThreadTask(List idList, String initiateTaskType, String userName) throws Exception {
        //具体的逻辑代码
        Map map = new HashMap();
        return map;
    }

方法二的执行结果,循环多少次就启动了多少个子线程,所以这里的想法是先将原生数组按自定义个进行分配,如有200个任务,分给20个人,每人10个大概就是这样的思路

java多线程带返回值的方式方法,在这里插入图片描述,第5张

java多线程带返回值的方式方法,在这里插入图片描述,第6张

我的完整代码仅供参考,里面很多类都是我自己业务用到的,大家可以借鉴

 public Map initiateTargetPriceTask(PmpTargetPriceDTO pmpTargetPriceDTO) throws Exception {
        String userName = SecurityUtils.getUsername();
        Map map = new HashMap();
        List list = Arrays.asList(pmpTargetPriceDTO.getIds());
        // 将要发送的集合按10个一组进行重新组装
        List> partition = Lists.partition(list, 10);
        // 创建一个线程池
        // 获取CPU核数
        int coresNumber = Runtime.getRuntime().availableProcessors();
        System.out.println("获取CPU核数:" + coresNumber);
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(coresNumber * 2, coresNumber * 2 + 1, 1000, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
        // 获取任务发起类型字段
        String initiateTaskType = pmpTargetPriceDTO.getInitiateTaskType();
        /*创建List用来接收多线程返回的内容,泛型里面可以自定义,String或者对象亦或者其他类型*/
        List> listret = new ArrayList<>();
        // 同步工具类,让主线程一直等待,直到子线程执行完后再执行
        CountDownLatch downLatch = new CountDownLatch(partition.size());
        // 循环任务的List
        for (List stringList : partition) {
            // 启用开启多个线程
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 开始调用具体业务代码
                        Map mapRet = pmpTargetPriceService.targetPriceThreadTask(stringList, initiateTaskType, userName);
                        listret.add(mapRet);
                    } catch (Exception e) {
                        logger.error("循环开启线多线程报错,调用下游系统出现错误,异常:" + e);
                    } finally {
                        // 业务逻辑处理完毕,计数器减一【当前线程处理任务完毕,线程释放进入线程池,等待处理下一个任务】
                        downLatch.countDown();
                    }
                }
            });
        }
        // 主线程需要等待子任务线程执行完,结果汇总之后,主线程继续往下执行
        try {
            downLatch.await();
        } catch (Exception e) {
            logger.error("等待超时", e);
            throw new RuntimeException("系统处理超时,请稍后再试");
        }
        // 关闭线程池
        executor.shutdown();
        // 获取所有并发任务的运行结果
        StringBuilder successMsg = new StringBuilder();
        StringBuilder failureMsg = new StringBuilder();
        int failureNum;
        int successNum;
        int failureNumCount = 0;
        int successNumCount = 0;
        for (Map esbResultPlm1 : listret) {
            // 从Future对象上获取任务的返回值,并输出到控制台
//            Map esbResultPlm1 = (Map) f.get();
            // todo 对我返回的多个map进行拼接
            if (esbResultPlm1.get("status").equals("fail")) {
                failureNum = (int) esbResultPlm1.get("failureNum");
                failureMsg.append(esbResultPlm1.get("msg"));
                map.put("msg", failureMsg.toString());
                failureNumCount += failureNum;
            } else {
                successNum = (int) esbResultPlm1.get("successNum");
                successMsg.append(esbResultPlm1.get("msg"));
                map.put("msg", successMsg.toString());
                successNumCount += successNum;
            }
        }
//        todo 对最终的结果进行组装
        if (failureNumCount > 0) {
            failureMsg.insert(0, "很抱歉,发起任务存在失败!共发起 " + list.size() + "条数据,其中有" + failureNumCount + " 条数据格式不正确,错误如下:");
            map.put("status", "fail");
            map.put("msg", failureMsg.toString());
        } else {
            successMsg.insert(0, "恭喜您,数据已全部发起成功!共 " + successNumCount + " 条");
            map.put("status", "success");
            map.put("msg", successMsg.toString());
        }
        return map;
    }
    
    // 调用的逻辑处理方法
	public Map targetPriceThreadTask(List idList, String initiateTaskType, String userName) throws Exception {
        // 发起目标价任务
        int successNum = 0;
        int failureNum = 0;
        StringBuilder successMsg = new StringBuilder();
        StringBuilder failureMsg = new StringBuilder();
        StringBuffer NoSubunitmaterialCode = new StringBuffer(); // 子组不存在的物料号合集
        StringBuffer NoSubunit = new StringBuffer(); // 没有子组的子组号合集
        Map map = new HashMap();
        for (String id : idList) {
            PmpTargetPrice pmpTargetPrice = pmpTargetPriceMapper.selectPmpTargetPriceById(id);
            SysApiRequestLog sysApiRequestLog = new SysApiRequestLog();
            sysApiRequestLog.setRequestMethod("手动发起目标价任务");
            sysApiRequestLog.setRequestData("物料号:" + pmpTargetPrice.getMaterialCode());
            //查询是否发起流程,
            if (pmpTargetPrice.getIsFqlc().equals("1")) {
                failureNum++;
                String msg = "
物料号:" + pmpTargetPrice.getMaterialCode() + "、此物料已经发起过流程,请核实!"; failureMsg.append(msg); continue; } PmpTargetPriceProcess targetPriceProcess = new PmpTargetPriceProcess(); // 请求PLM接口 Map invokeGetPlm = invokeWebService.getInvokeGetPlm(pmpTargetPrice.getMaterialCode()); targetPriceProcess.setSouce("手动发起"); // 来源(手动发起) targetPriceProcess.setTaskSponsor(userName); // 设置发起人 targetPriceProcess.setMaterialStatus("0"); // 状态 targetPriceProcess.setInitiateTaskType(initiateTaskType); // 设置手工发起任务类型 if (null != invokeGetPlm.get("number").toString()) { targetPriceProcess.setMaterialCode(invokeGetPlm.get("number").toString()); // 物料编号 } if (null != invokeGetPlm.get("name").toString()) { targetPriceProcess.setMaterialName(invokeGetPlm.get("name").toString()); // 物料名称 } if (null != invokeGetPlm.get("phase").toString()) { targetPriceProcess.setStage(invokeGetPlm.get("phase").toString());//阶段 } if (null != invokeGetPlm.get("version").toString()) { targetPriceProcess.setVersionNo(invokeGetPlm.get("version").toString()); // 大版本 } if (null != invokeGetPlm.get("state").toString()) { targetPriceProcess.setMaterialStatus(invokeGetPlm.get("state").toString()); // 状态 } // 请求BOM接口获取数据 Map materialCode = invokeWebService.getEsbBomMaterialInfo(pmpTargetPrice.getMaterialCode()); // 截取物料编码为子组号 String substring = pmpTargetPriceProcessService.getBlockCode(pmpTargetPrice.getMaterialCode()); PmpTargetRule targetRule = new PmpTargetRule(); PmpTargetRule pmpTargetRule; String userCode = ""; // 判断bom是否有返回,有返回表示有路线,无返回表示无路线 if (null != materialCode) { //根据物料编号截取的子组取目标规则表PMP_TARGET_RULE中查 targetRule.setSonGroup(substring); targetRule.setIsRoute("1"); pmpTargetRule = pmpTargetRuleService.selectPmpTargetRuleBySonGroup(targetRule); // 判断子组配置是否存在,如根据子组查询不存在则设置为特殊子组 if (!Optional.ofNullable(pmpTargetRule).isPresent()) { targetRule.setSonGroup("特殊件无法获取"); targetRule.setIsRoute("1"); pmpTargetRule = pmpTargetRuleService.selectPmpTargetRuleBySonGroup(targetRule); } /** * 1.一级制造、一级装配和采购制造均为空时生成任务; * 2.一级制造、一级装配和采购制造均有值时生成任务; * 3.一级制造、一级装配有值,采购制造为空时,不生成任务。但要判定一级制造和一级装配均不含CG时为自制件, * 不生成任务,但返回PLM为S,提示“一级制造和一级装配均不含CG,为自制件,不生成任务”, * 其他情况返回PLM为E并提示“非自制件,BOM采购制造路线待维护,请稍后发起定价任务”。 */ String mfmrtg = ""; // 一级制造 String mfartg = ""; // 一级装配 if (StringUtils.isNotBlank(materialCode.get("MFMRTG").toString()) && StringUtils.isNotBlank(materialCode.get("MFARTG").toString()) ) { mfmrtg = materialCode.get("MFMRTG").toString(); // 一级制造 mfartg = materialCode.get("MFARTG").toString(); // 一级装配 } else { failureNum++; logger.error("物料号:" + pmpTargetPrice.getMaterialCode() + "获取BOM信息,一级制造或一级装配为Null,发起任务失败!"); String msg = "
物料号:" + pmpTargetPrice.getMaterialCode() + "、获取BOM信息,一级制造或一级装配为Null,发起任务失败! "; failureMsg.append(msg); sysApiRequestLog.setResponseStatus(EsbResult.RET_STATUS_ERROR); sysApiRequestLog.setErrorLog("手动发起目标价任务,失败原因:根据物料号获取BOM信息,一级制造或一级装配为Null,发起任务失败"); apiRequestLogService.insertSysApiRequestLog(sysApiRequestLog); continue; } // if (StringUtils.isNotBlank(materialCode.get("MFMRTG").toString())) { // targetPriceProcess.setOneLevelMake(materialCode.get("MFMRTG").toString()); // 一级制造 // } // if (StringUtils.isNotBlank(materialCode.get("MFARTG").toString())) { // targetPriceProcess.setOneLevelAssembling(materialCode.get("MFARTG").toString()); // 一级装配 // } targetPriceProcess.setOneLevelMake(mfmrtg); // 一级制造 targetPriceProcess.setOneLevelAssembling(mfartg); // 一级装配 if (StringUtils.isNotBlank(materialCode.get("CFMRTG").toString())) { targetPriceProcess.setPurchaseMake(materialCode.get("CFMRTG").toString()); // 采购制造 //根据BOM接口返回采购制造,如果是PT为研究院,否则为财务部 if (targetPriceProcess.getPurchaseMake().equals("PT")) { if (!StringUtils.isEmpty(pmpTargetRule.getYjyDirectorCode())) { targetPriceProcess.setTaskPurchase(pmpTargetRule.getYjyDirectorCode()); // 指定填写目标价人:研究院 // 设置审核人 String[] splitYjyCode = pmpTargetRule.getYjyDirectorCode().split("/"); userCode = splitYjyCode[0]; } else { failureNum++; String msg = "
物料号:" + pmpTargetPrice.getMaterialCode() + "、未获取到研究院相关人员,请核对! "; failureMsg.append(msg); logger.error("手动发起目标价任务-物料号-PT-YJY:" + pmpTargetPrice.getMaterialCode() + "设置目标价录入人时配置有误,路线为PT,根据子组查询但是研究院人员code是未维护发起任务失败!"); sysApiRequestLog.setResponseStatus(EsbResult.RET_STATUS_ERROR); sysApiRequestLog.setErrorLog("手动发起目标价任务失败,失败原因:路线为PT,根据子组查询但是研究院人员code是未维护发起任务失败"); apiRequestLogService.insertSysApiRequestLog(sysApiRequestLog); NoSubunitmaterialCode.append("物料号-PT-YJY:" + pmpTargetPrice.getMaterialCode() + "/"); NoSubunit.append("子组号-PT-YJY:" + substring + "/"); continue; } } else { if (!StringUtils.isEmpty(pmpTargetRule.getCwDirectorCode())) { targetPriceProcess.setTaskPurchase(pmpTargetRule.getCwDirectorCode()); // 指定填写目标价人:财务部 // 设置审核人 String[] splitCwCode = pmpTargetRule.getCwDirectorCode().split("/"); userCode = splitCwCode[0]; } else { failureNum++; String msg = "
物料号:" + pmpTargetPrice.getMaterialCode() + "、未获取到财务部相关人员,请核对! "; failureMsg.append(msg); logger.error("手动发起目标价任务-物料号-非PT-CW:" + pmpTargetPrice.getMaterialCode() + "设置目标价录入人时配置有误,路线非PT,根据子组查询财务人员code是Null发起任务失败!"); sysApiRequestLog.setResponseStatus(EsbResult.RET_STATUS_ERROR); sysApiRequestLog.setErrorLog("手动发起目标价任务,失败原因:根据子组查询财务人员code是Null发起任务失败"); apiRequestLogService.insertSysApiRequestLog(sysApiRequestLog); NoSubunitmaterialCode.append("物料号-非PT-CW:" + pmpTargetPrice.getMaterialCode() + "/"); NoSubunit.append("子组号-非PT-CW:" + substring + "/"); continue; } } } else { if (!mfmrtg.equals("CG") || !mfartg.equals("CG")) { logger.error("物料号:" + pmpTargetPrice.getMaterialCode() + "一级制造和一级装配均不含CG,为自制件,不生成任务!"); failureNum++; String msg = "
物料号:" + pmpTargetPrice.getMaterialCode() + "、一级制造和一级装配均不含CG,为自制件,不生成任务! "; failureMsg.append(msg); sysApiRequestLog.setResponseStatus(EsbResult.RET_STATUS_ERROR); sysApiRequestLog.setErrorLog("手动发起目标价任务,失败原因:一级制造和一级装配均不含CG,为自制件,不生成任务!"); continue; } else { failureNum++; String msg = "
物料号:" + pmpTargetPrice.getMaterialCode() + "、BOM采购制造路线待维护,请稍后发起定价任务!"; failureMsg.append(msg); sysApiRequestLog.setResponseStatus(EsbResult.RET_STATUS_ERROR); sysApiRequestLog.setErrorLog("手动发起目标价任务,失败原因:BOM采购制造路线待维护,请稍后发起定价任务!"); continue; } } } else { // 无路线 targetRule.setSonGroup(substring); targetRule.setIsRoute("0"); pmpTargetRule = pmpTargetRuleService.selectPmpTargetRuleBySonGroup(targetRule); /** * 其他特殊件,无法获取子组, * PT或无路线由研究院郑宇处理, * 非PT的由财务部苏战波和赵伟处理 */ if (!Optional.ofNullable(pmpTargetRule).isPresent()) { targetRule.setSonGroup("特殊件无法获取"); targetRule.setIsRoute("1"); pmpTargetRule = pmpTargetRuleService.selectPmpTargetRuleBySonGroup(targetRule); } if (null != pmpTargetRule && StringUtils.isNotBlank(pmpTargetRule.getYjyDirectorCode())) { targetPriceProcess.setTaskPurchase(pmpTargetRule.getYjyDirectorCode()); // 指定填写目标价人:研究院 String[] splitYjyCode = pmpTargetRule.getYjyDirectorCode().split("/"); userCode = splitYjyCode[0]; } else { failureNum++; String msg = "
物料号:" + pmpTargetPrice.getMaterialCode() + "、未获取到研究院相关人员(无路线),请核对! "; failureMsg.append(msg); continue; } } // todo 判断任务是研究院或者财务,设置审核人 // 查询研究院,财务或者无路线任务办理人,并获取他们的部门编码,向上寻找审核人 if (!userCode.equals("")) { SysUser sysUser = userService.selectUserByUserName(userCode); if (StringUtils.isNotNull(sysUser)) { // 截取部门code String deptCode = ""; if (sysUser.getDeptId().length() > 9) { deptCode = sysUser.getDeptId().substring(0, 9); } else { deptCode = sysUser.getDeptId(); } //查询审核配置表,查到审核人,插入目标价 PmpTargetAuditConfig pmpTargetAuditConfig = pmpTargetAuditConfigService.selectPmpTargetAuditConfigByDeptCode(deptCode); targetPriceProcess.setCheckName(pmpTargetAuditConfig.getAuditor()); } } //已发起流程 pmpTargetPrice.setIsFqlc("1"); targetPriceProcess.setStatus("0"); // 任务状态 0待定级发起 targetPriceProcess.setQuejiaType("目标价"); // 缺价类型默认目标价 //添加任务发起人 targetPriceProcess.setTaskSponsor(userName); targetPriceProcess.setCreateBy(userName); pmpTargetPriceMapper.updatePmpTargetPrice(pmpTargetPrice); if (StringUtils.isEmpty(targetPriceProcess.getMaterialName())) { targetPriceProcess.setMaterialName(pmpTargetPrice.getMaterialName()); } /** * 查看当前物料号在目标价任务表中是否存在,最后的检查 */ PmpTargetPriceProcess priceProcess = new PmpTargetPriceProcess(); priceProcess.setMaterialCode(targetPriceProcess.getMaterialCode()); priceProcess.setStatus("0,1,2,4"); PmpTargetPriceProcess pmpTargetPriceProcess = targetPriceProcessService.selectPmpTargetPriceProcessByEntity(priceProcess); if (null != pmpTargetPriceProcess) { continue; } // 设置任务号 String newsNo = DateUtils.parseDateToStr("yyyyMMdd", new Date()); int count = targetPriceProcessService.getFindTaskCount(); String format = String.format("%05d", count); // 任务编号 MBJRW+年月日+流水号 targetPriceProcess.setTaskNumber("MBJRW" + newsNo + format); targetPriceProcessService.insertPmpTargetPriceProcess(targetPriceProcess); successNum++; } if (failureNum > 0) { // failureMsg.insert(0, "很抱歉,发起任务存在失败!共 " + failureNum + " 条数据格式不正确,错误如下:"); map.put("status", "fail"); map.put("msg", failureMsg.toString()); map.put("failureNum", failureNum); } else { // successMsg.insert(0, "恭喜您,数据已全部发起成功!共 " + successNum + " 条"); map.put("status", "success"); map.put("msg", successMsg.toString()); map.put("successNum", successNum); } logger.info("手动发起目标价任务结束!"); logger.info("手动发起目标价-物料号查找子组有误的物料统计:" + NoSubunitmaterialCode.toString()); logger.info("手动发起目标价-物料号查找子组有误的子组号统计:" + NoSubunit.toString()); return map; }

总结方式一和方式二都能解决加快任务处理,处理时间都差不读多,大家可以挑选自己适合的方式,如有更好的方式或不对的点请指正,欢迎大家沟通交流,共同成长进步