确认导入多线程处理

tnc-sms/src/main/java/org/jeecg/modules/smscheck/controller/SmsCheckTaskController.java

@@ -15,7 +15,6 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.shiro.SecurityUtils;
 import org.jeecg.common.api.vo.Result;
 import org.jeecg.common.aspect.annotation.AutoLog;
 import org.jeecg.common.aspect.annotation.PermissionData;
@@ -24,7 +23,6 @@ import org.jeecg.common.system.api.ISysBaseAPI;
 import org.jeecg.common.system.base.controller.JeecgController;
 import org.jeecg.common.system.query.QueryGenerator;
 import org.jeecg.common.system.vo.DictModel;
-import org.jeecg.common.system.vo.LoginUser;
 import org.jeecg.common.system.vo.SysDepartModel;
 import org.jeecg.modules.smscheck.configs.CheckStateCode;
 import org.jeecg.modules.smscheck.entity.*;
@@ -44,6 +42,7 @@ import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -376,12 +375,13 @@ public class SmsCheckTaskController extends JeecgController<SmsCheckTask, ISmsCh
      * @return 导入结果
      */
     @PostMapping(value = "/{taskId}/importCustomerData")
-    public Result<?> importCustomerData(@PathVariable String taskId, @RequestBody List<SmsCheckCustomerInfoVo> list) {
+    public Result<?> importCustomerData(@PathVariable String taskId, @RequestBody List<SmsCheckCustomerInfoVo> list) throws ExecutionException, InterruptedException {
         UUID uuid = UUID.randomUUID();
         //1同一个任务 用户编号不可一致
         //2 员工得存在
-        LoginUser sysUser = (LoginUser) SecurityUtils.getSubject().getPrincipal();
-        List<SmsCheckCustomerInfo> dataList = new ArrayList<>();
+        // 开始时间
+        long start = System.currentTimeMillis();
+       // List<SmsCheckCustomerInfo> dataList = new ArrayList<>();
         //导入失败数据
         List<SmsCheckCustomerInfoVo> listErro = new ArrayList<>();
         int customerCount = 0;
@@ -392,135 +392,55 @@ public class SmsCheckTaskController extends JeecgController<SmsCheckTask, ISmsCh
         Map<String, SysUser> userMap
                 = userList.stream().collect(Collectors.toMap(SysUser::getWorkNo, Function.identity(), (o, n) -> o));
 
-        for (int i = 0; i < list.size(); i++) {
-            SmsCheckCustomerInfoVo infoVo = list.get(i);
-            infoVo.setBatchNo(uuid.toString());
-            infoVo.setTaskId(taskId);
-            SmsCheckCustomerInfo info = new SmsCheckCustomerInfo();
-            BeanUtil.copyProperties(infoVo, info, true);
-
-//            SysUser user = sysUserService.getOne(new LambdaQueryWrapper<SysUser>()
-//                    .eq(SysUser::getWorkNo, infoVo.getStaffNo()));
-            //放入用户信息
-            SysUser user =userMap.get( infoVo.getStaffNo());
-            if (user == null) {
-                infoVo.setResult("0");
-                infoVo.setErrorMsg("系统无此员工");
-                listErro.add(infoVo);
-                continue;
-            }
-            info.setStaffUsername(user.getUsername());
-            if (info.getSysOrgCode()!=null){
-                // 转派 可以根据页面传 orgCOde进行 判断    撤回不可以 所以在这让撤回只走 原逻辑    去掉 if (info.getSysOrgCode()!=null){
-                if (info.getSysOrgCode()==null){
-                    infoVo.setResult("0");
-                    infoVo.setErrorMsg("机构未查到");
-                    listErro.add(infoVo);
-                    continue;
-                }
-                List<String> deptNameByUserId = taskService.getDeptNameByUserId(user.getWorkNo());
-                if (!deptNameByUserId.contains(info.getSysOrgCode())){
-                    infoVo.setResult("0");
-                    infoVo.setErrorMsg("此员工机构选择错误");
-                    listErro.add(infoVo);
-                    continue;
-                }
-            }
-            if (info.getStaffDeptLevel2Name()==null || info.getStaffDeptLevel2No()==null) {
-                infoVo.setResult("0");
-                infoVo.setErrorMsg("二级部门信息为空");
-                listErro.add(infoVo);
-                continue;
-            }
-            if (info.getStaffDeptLevel3Name()==null || info.getStaffDeptLevel3No()==null) {
-                infoVo.setResult("0");
-                infoVo.setErrorMsg("三级部门信息为空");
-                listErro.add(infoVo);
-                continue;
-            }
+        // 每20条数据开启一条线程
+        int threadSize = 20;
+        // 总数据条数
+        int dataSize = list.size();
+        // 线程数
+        int threadNum = dataSize / threadSize + 1;
+        //暂定20个一组
+        boolean special = list.size() % threadSize == 0;
 
-            info.setDelFlag(CommonConstant.DEL_FLAG_0);
-            // 0：未处理；1：未完善；2：待稽核；3：待整改；4：已整改；5：稽核通过 6 未上传；
-            info.setCheckState(CheckStateCode.CHECK_STSTE_0);
-            dataList.add(info);
-            info.setTaskId(taskId);
-            long count = customerInfoService.count(
-                    new LambdaQueryWrapper<SmsCheckCustomerInfo>()
-                            .eq(SmsCheckCustomerInfo::getTaskId, info.getTaskId())
-                            .eq(SmsCheckCustomerInfo::getUserNo, info.getUserNo())
-                            .eq(SmsCheckCustomerInfo::getDelFlag, CommonConstant.DEL_FLAG_0));
-
-            long counted = customerInfoService.count(
-                    new LambdaQueryWrapper<SmsCheckCustomerInfo>()
-                            .eq(SmsCheckCustomerInfo::getTaskId, info.getTaskId())
-                            .eq(SmsCheckCustomerInfo::getCustomerNo, info.getCustomerNo())
-                            .eq(SmsCheckCustomerInfo::getDelFlag, CommonConstant.DEL_FLAG_0));
-
-            if (count > 0 && counted > 0) {
-                infoVo.setResult("0");
-                infoVo.setErrorMsg("用户编号重复,客户编号重复");
-                listErro.add(infoVo);
-            }else {
-                if (count > 0 && counted < 0) {
-                    infoVo.setResult("0");
-                    infoVo.setErrorMsg("用户编号重复");
-                    listErro.add(infoVo);
-                    // TODO 重复处理，放校验里面？
-//				 return Result.error("客户[" + info.getCustomerNo() + "]已存在！");
-                } else if (count < 0 && counted > 0) {
-                    infoVo.setResult("0");
-                    infoVo.setErrorMsg("客户编号重复");
-                    listErro.add(infoVo);
-                    // TODO 重复处理，放校验里面？
-//				 return Result.error("客户[" + info.getCustomerNo() + "]已存在！");
-                } else {
-                    customerCount++;
-                    customerInfoService.save(info);
-                    infoVo.setResult("1");
-                    listErro.add(infoVo);
+        ExecutorService executorService = new ThreadPoolExecutor(threadNum, threadNum+5,
+                0L, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<Runnable>(10) );
+
+        Map<String,List<SmsCheckCustomerInfoVo>> mapData = new HashMap<>();
+        List<SmsCheckCustomerInfoVo> cutList ;
+        for (int i = 0; i < threadNum; i++) {
+            //最后一次
+            if (i == threadNum - 1) {
+                if (special) {
+                    break;
                 }
-            }
 
-            // 添加公司信息 start
-            long companyCount = companyService.count(new LambdaQueryWrapper<SmsCheckCompany>()
-                    .eq(SmsCheckCompany::getCustomerNo, infoVo.getCustomerNo())
-                    .eq(SmsCheckCompany::getCustomerName, infoVo.getCustomerName()));
-            if (companyCount == 0) {
-                SmsCheckCompany company = new SmsCheckCompany();
-                company.setCustomerNo(infoVo.getCustomerNo());
-                company.setCustomerName(infoVo.getCustomerName());
-                company.setManager(infoVo.getStaffName());
-                company.setSysOrgCode(sysUser.getOrgCode());
-                company.setCreateBy(sysUser.getUsername());
-                company.setCreateTime(new Date());
-                companyService.save(company);
-            }
-            // 添加公司信息 end
-
-            // 添加服务号码 start
-            long smsNumberCount = smsNumberService.count(new LambdaQueryWrapper<SmsCheckSmsNumber>()
-                    .eq(SmsCheckSmsNumber::getCustomerNo, infoVo.getCustomerNo())
-                    .eq(SmsCheckSmsNumber::getSmsNumber, infoVo.getSmsNumber()));
-
-            if (smsNumberCount == 0) {
-                SmsCheckSmsNumber smsNumber = new SmsCheckSmsNumber();
-                smsNumber.setCustomerNo(infoVo.getCustomerNo());
-                smsNumber.setUserNo(infoVo.getUserNo());
-                smsNumber.setNetworkAccessTime(infoVo.getNetworkAccessTime());
-                smsNumber.setSmsNumber(infoVo.getSmsNumber());
-                smsNumber.setUserState(infoVo.getUserState());
-                smsNumber.setSysOrgCode(sysUser.getOrgCode());
-                smsNumber.setCreateBy(sysUser.getUsername());
-                smsNumber.setCreateTime(new Date());
-                smsNumberService.save(smsNumber);
+                mapData.put(""+i, list.subList( threadSize * i, list.size()));
+            } else {
+                mapData.put(""+i,list.subList(  threadSize * i,  threadSize * (i + 1)));
             }
-            // 添加服务号码 end
-
+        }
+        AtomicInteger compareNum = new AtomicInteger(0);
+        for (String key : mapData.keySet()) {
+            List<SmsCheckCustomerInfoVo> smsCheckCustomerInfoVos = mapData.get(key);
+            ImportCustomerDataTask importCustomerDataTask = new ImportCustomerDataTask();
+            importCustomerDataTask.setUuid(uuid);
+            importCustomerDataTask.setTaskId(taskId);
+            importCustomerDataTask.setUserMap(userMap);
+            importCustomerDataTask.setList(smsCheckCustomerInfoVos);
+            importCustomerDataTask.setTaskService(taskService);
+            importCustomerDataTask.setCompanyService(companyService);
+            importCustomerDataTask.setSmsNumberService(smsNumberService);
+            importCustomerDataTask.setCompareNum(compareNum);
+            importCustomerDataTask.setCustomerInfoService(customerInfoService);
+
+            Future submit = executorService.submit(importCustomerDataTask);
+            List<SmsCheckCustomerInfoVo>  o = (List<SmsCheckCustomerInfoVo>) submit.get();
+            listErro.addAll(o);
         }
         SmsCheckTask task = taskService.getById(taskId);
         // 更新导入
-        if (customerCount > 0) {
-            task.setCustomerCount(task.getCustomerCount() + customerCount);
+        if (list.size()-compareNum.get()>0) {
+            task.setCustomerCount(task.getCustomerCount() + (list.size()-compareNum.get()));
             taskService.updateById(task);
         }
         if (listErro!=null && listErro.size()>0){
@@ -536,9 +456,10 @@ public class SmsCheckTaskController extends JeecgController<SmsCheckTask, ISmsCh
 
             }
             boolean b = smsCheckCustomerInfoErrorService.saveOrUpdateBatch(smsCheckCustomerInfoErrors);
-            int i=0;
         }
-        return Result.OK("导入成功！导入成功条数" + customerCount+"。 导入失败条数"+(list.size()-customerCount), listErro);
+        executorService.shutdown();
+        System.err.println("执行任务消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
+        return Result.OK("导入成功！导入成功条数" + ( listErro.size()-compareNum.get())+"。 导入失败条数"+ compareNum , listErro);
     }
 
     /**

tnc-sms/src/main/java/org/jeecg/modules/smscheck/entity/ImportCustomerDataTask.java

@@ -0,0 +1,202 @@
+package org.jeecg.modules.smscheck.entity;
+
+import cn.hutool.core.bean.BeanUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import lombok.Data;
+import org.apache.shiro.SecurityUtils;
+import org.jeecg.common.constant.CommonConstant;
+import org.jeecg.common.system.api.ISysBaseAPI;
+import org.jeecg.common.system.vo.LoginUser;
+import org.jeecg.modules.smscheck.configs.CheckStateCode;
+import org.jeecg.modules.smscheck.service.*;
+import org.jeecg.modules.smscheck.service.impl.DataYearMonthImpl;
+import org.jeecg.modules.smscheck.vo.SmsCheckCustomerInfoVo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.*;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicInteger;
+
+
+@Data
+public class ImportCustomerDataTask implements Callable {
+
+    /**
+     * 计数器
+     */
+    private   AtomicInteger compareNum;
+    
+    private ISmsCheckTaskService taskService;
+    
+    private ISmsCheckCustomerInfoService customerInfoService;
+    
+    private ISmsCheckCompanyService companyService;
+    
+    private ISmsCheckSmsNumberService smsNumberService;
+    
+    private ISysBaseAPI sysBaseApi;
+
+
+    
+    private ISysUserService sysUserService;
+
+    
+    private DataYearMonthImpl dataYearMonth;
+    
+    private ISmsCheckCustomerInfoErrorService smsCheckCustomerInfoErrorService;
+
+    private Map<String, SysUser> userMap ;
+
+    private List<SmsCheckCustomerInfoVo> list ;
+
+    private UUID uuid;
+
+    private String taskId;
+
+    @Override
+    public List<SmsCheckCustomerInfoVo> call() throws Exception {
+        return  test();
+    }
+
+    public List<SmsCheckCustomerInfoVo>  test(){
+
+        List<SmsCheckCustomerInfoVo> listErro = new ArrayList<>();
+
+        LoginUser sysUser = (LoginUser) SecurityUtils.getSubject().getPrincipal();
+        for (int i = 0; i < list.size(); i++) {
+            SmsCheckCustomerInfoVo infoVo = list.get(i);
+            infoVo.setBatchNo(uuid.toString());
+            infoVo.setTaskId(taskId);
+            SmsCheckCustomerInfo info = new SmsCheckCustomerInfo();
+            BeanUtil.copyProperties(infoVo, info, true);
+
+//            SysUser user = sysUserService.getOne(new LambdaQueryWrapper<SysUser>()
+//                    .eq(SysUser::getWorkNo, infoVo.getStaffNo()));
+            //放入用户信息
+            SysUser user =userMap.get( infoVo.getStaffNo());
+            if (user == null) {
+                infoVo.setResult("0");
+                infoVo.setErrorMsg("系统无此员工");
+                listErro.add(infoVo);
+                compareNum.getAndIncrement();
+                continue;
+            }
+            info.setStaffUsername(user.getUsername());
+            if (info.getSysOrgCode()!=null){
+                // 转派 可以根据页面传 orgCOde进行 判断    撤回不可以 所以在这让撤回只走 原逻辑    去掉 if (info.getSysOrgCode()!=null){
+                if (info.getSysOrgCode()==null){
+                    infoVo.setResult("0");
+                    infoVo.setErrorMsg("机构未查到");
+                    compareNum.getAndIncrement();
+                    listErro.add(infoVo);
+                    continue;
+                }
+                List<String> deptNameByUserId = taskService.getDeptNameByUserId(user.getWorkNo());
+                if (!deptNameByUserId.contains(info.getSysOrgCode())){
+                    infoVo.setResult("0");
+                    infoVo.setErrorMsg("此员工机构选择错误");
+                    compareNum.getAndIncrement();
+                    listErro.add(infoVo);
+                    continue;
+                }
+            }
+            if (info.getStaffDeptLevel2Name()==null || info.getStaffDeptLevel2No()==null) {
+                infoVo.setResult("0");
+                infoVo.setErrorMsg("二级部门信息为空");
+                compareNum.getAndIncrement();
+                listErro.add(infoVo);
+                continue;
+            }
+            if (info.getStaffDeptLevel3Name()==null || info.getStaffDeptLevel3No()==null) {
+                infoVo.setResult("0");
+                infoVo.setErrorMsg("三级部门信息为空");
+                compareNum.getAndIncrement();
+                listErro.add(infoVo);
+                continue;
+            }
+
+            info.setDelFlag(CommonConstant.DEL_FLAG_0);
+            // 0：未处理；1：未完善；2：待稽核；3：待整改；4：已整改；5：稽核通过 6 未上传；
+            info.setCheckState(CheckStateCode.CHECK_STSTE_0);
+            //dataList.add(info);
+            info.setTaskId(taskId);
+            long count = customerInfoService.count(
+                    new LambdaQueryWrapper<SmsCheckCustomerInfo>()
+                            .eq(SmsCheckCustomerInfo::getTaskId, info.getTaskId())
+                            .eq(SmsCheckCustomerInfo::getUserNo, info.getUserNo())
+                            .eq(SmsCheckCustomerInfo::getDelFlag, CommonConstant.DEL_FLAG_0));
+
+            long counted = customerInfoService.count(
+                    new LambdaQueryWrapper<SmsCheckCustomerInfo>()
+                            .eq(SmsCheckCustomerInfo::getTaskId, info.getTaskId())
+                            .eq(SmsCheckCustomerInfo::getCustomerNo, info.getCustomerNo())
+                            .eq(SmsCheckCustomerInfo::getDelFlag, CommonConstant.DEL_FLAG_0));
+
+            if (count > 0 && counted > 0) {
+                infoVo.setResult("0");
+                compareNum.getAndIncrement();
+                infoVo.setErrorMsg("用户编号重复,客户编号重复");
+                listErro.add(infoVo);
+            }else {
+                if (count > 0 && counted < 0) {
+                    infoVo.setResult("0");
+                    compareNum.getAndIncrement();
+                    infoVo.setErrorMsg("用户编号重复");
+                    listErro.add(infoVo);
+                    // TODO 重复处理，放校验里面？
+//				 return Result.error("客户[" + info.getCustomerNo() + "]已存在！");
+                } else if (count < 0 && counted > 0) {
+                    infoVo.setResult("0");
+                    compareNum.getAndIncrement();
+                    infoVo.setErrorMsg("客户编号重复");
+                    listErro.add(infoVo);
+                    // TODO 重复处理，放校验里面？
+//				 return Result.error("客户[" + info.getCustomerNo() + "]已存在！");
+                } else {
+                    //customerCount++;
+                    customerInfoService.save(info);
+                    infoVo.setResult("1");
+                    listErro.add(infoVo);
+                }
+            }
+
+            // 添加公司信息 start
+            long companyCount = companyService.count(new LambdaQueryWrapper<SmsCheckCompany>()
+                    .eq(SmsCheckCompany::getCustomerNo, infoVo.getCustomerNo())
+                    .eq(SmsCheckCompany::getCustomerName, infoVo.getCustomerName()));
+            if (companyCount == 0) {
+                SmsCheckCompany company = new SmsCheckCompany();
+                company.setCustomerNo(infoVo.getCustomerNo());
+                company.setCustomerName(infoVo.getCustomerName());
+                company.setManager(infoVo.getStaffName());
+                company.setSysOrgCode(sysUser.getOrgCode());
+                company.setCreateBy(sysUser.getUsername());
+                company.setCreateTime(new Date());
+                companyService.save(company);
+            }
+            // 添加公司信息 end
+
+            // 添加服务号码 start
+            long smsNumberCount = smsNumberService.count(new LambdaQueryWrapper<SmsCheckSmsNumber>()
+                    .eq(SmsCheckSmsNumber::getCustomerNo, infoVo.getCustomerNo())
+                    .eq(SmsCheckSmsNumber::getSmsNumber, infoVo.getSmsNumber()));
+
+            if (smsNumberCount == 0) {
+                SmsCheckSmsNumber smsNumber = new SmsCheckSmsNumber();
+                smsNumber.setCustomerNo(infoVo.getCustomerNo());
+                smsNumber.setUserNo(infoVo.getUserNo());
+                smsNumber.setNetworkAccessTime(infoVo.getNetworkAccessTime());
+                smsNumber.setSmsNumber(infoVo.getSmsNumber());
+                smsNumber.setUserState(infoVo.getUserState());
+                smsNumber.setSysOrgCode(sysUser.getOrgCode());
+                smsNumber.setCreateBy(sysUser.getUsername());
+                smsNumber.setCreateTime(new Date());
+                smsNumberService.save(smsNumber);
+            }
+            // 添加服务号码 end
+
+        }
+        return listErro;
+    }
+}