同步数据增加线程池

src/main/java/common/model/TkioFlowHistory.java

@@ -21,6 +21,7 @@ public class TkioFlowHistory {
     private String yMonth;
     private String yMonthNext;
     private String nextCode;
+    private Integer firstRecode;
 
 
 
@@ -108,4 +109,12 @@ public class TkioFlowHistory {
     public void setNextCode(String nextCode) {
         this.nextCode = nextCode;
     }
+
+    public Integer getFirstRecode() {
+        return firstRecode;
+    }
+
+    public void setFirstRecode(Integer firstRecode) {
+        this.firstRecode = firstRecode;
+    }
 }

src/main/java/common/repository/CalculationFlowRepository.java

@@ -2,14 +2,20 @@ package common.repository;
 
 import common.model.CalculationFlow;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 
 @Transactional
-public interface CalculationFlowRepository extends JpaRepository<CalculationFlow,Long> {
+public interface CalculationFlowRepository extends JpaRepository<CalculationFlow, Long> {
 
-    @Query(value = "select * from calculation_flow where status = ?1",nativeQuery = true)
+    @Query(value = "select * from calculation_flow where status = ?1", nativeQuery = true)
     List<CalculationFlow> findByStatus(int status);
+
+    @Query(value = "update calculation_flow set status  = ?2 where id = ?1 ", nativeQuery = true)
+    @Transactional
+    @Modifying
+    void updateStatus(Long id, int status);
 }

src/main/java/track/task/TrackingFlowTask.java

@@ -24,6 +24,9 @@ import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.stream.Collectors;
 
 /**
@@ -52,16 +55,42 @@ public class TrackingFlowTask {
     @Autowired
     private TkioFlowHistoryRepository tkioFlowHistoryRepository;
 
+    private final ExecutorService executor = Executors.newFixedThreadPool(10);
+
+
     public void syncFlow() {
+
         List<CalculationFlow> calculationFlows = calculationFlowRepository.findByStatus(0);
         for (CalculationFlow calculationFlow : calculationFlows) {
-            List<TkioFlow> tkioFlowList = new ArrayList<>();
+            //List<TkioFlow> tkioFlowList = new ArrayList<>();
             calculationFlow.setStatus(1);
             calculationFlowRepository.save(calculationFlow);
             tkioFlowRepository.deleteByEmail(calculationFlow.getEmail());
             tkioFlowHistoryRepository.deleteByEmail(calculationFlow.getEmail());
             String email = calculationFlow.getEmail();
             //查询用户下所有appkey
+            //每个用户数据在单独线程处理
+            executor.execute(new RunableTask(email, calculationFlow));
+        }
+    }
+
+    private class RunableTask implements Runnable {
+        private String email;
+        private CalculationFlow calculationFlow;
+
+        public RunableTask(String email, CalculationFlow calculationFlow) {
+            this.email = email;
+            this.calculationFlow = calculationFlow;
+        }
+
+        @Override
+        public void run() {
+            calculationFlowByAccount(calculationFlow, email);
+        }
+
+    }
+
+    private void calculationFlowByAccount(CalculationFlow calculationFlow, String email) {
         try {
             Account account = accountRepository.findByEmail(email);
             List<Account> accountList = accountRepository.findByRootParent(account.getRootParent());
@@ -75,9 +104,10 @@ public class TrackingFlowTask {
 
             String ago = DateUtil.format(account.getCreateTime(), DateUtil.C_DATE_PATTON_DEFAULT);//查找最早一天的流量
             if (ago == null) {
-                    calculationFlow.setStatus(2);
-                    calculationFlowRepository.save(calculationFlow);
-                    continue;
+                /*calculationFlow.setStatus(2);
+                calculationFlowRepository.save(calculationFlow);*/
+                calculationFlowRepository.updateStatus(calculationFlow.getId(), 2);
+                return;
             }
 
             int between = 0;
@@ -98,7 +128,7 @@ public class TrackingFlowTask {
                     if (contracts.size() == 1) {//只有一个合同
                         TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contracts.get(0));
                         if (tkioFlow != null) {
-                                tkioFlowList.add(tkioFlow);
+                            //tkioFlowList.add(tkioFlow);
                             tkioFlow.setContractCount(1);
                             tkioFlowRepository.save(tkioFlow);
                         }
@@ -106,7 +136,6 @@ public class TrackingFlowTask {
 
                         getFlowOnMonths(email, yesterday, clickNum, contracts);//月流量报表逻辑
 
-
                         //多个合同
                         //看昨日被哪几个合同包含了
                         //昨日有几个合同开始生效
@@ -143,12 +172,12 @@ public class TrackingFlowTask {
                                     clickNum = BigInteger.valueOf(tkioFlow.getCostFlow());
                                     tkioFlow.setCostFlow(null);
                                     if (tkioFlow.getFlow() > 0) {
-                                            tkioFlowList.add(tkioFlow);
+                                        //tkioFlowList.add(tkioFlow);
                                         tkioFlowRepository.save(tkioFlow);
                                     }
                                 } else {
                                     if (tkioFlow != null) {
-                                            tkioFlowList.add(tkioFlow);
+                                        //tkioFlowList.add(tkioFlow);
                                         tkioFlowRepository.save(tkioFlow);
                                     }
                                     break;
@@ -177,7 +206,7 @@ public class TrackingFlowTask {
                                         TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[j]);
                                         tkioFlow.setContractCount(contracts.size());
                                         if (tkioFlow != null) {
-                                                tkioFlowList.add(tkioFlow);
+                                            //tkioFlowList.add(tkioFlow);
                                             tkioFlowRepository.save(tkioFlow);
                                         }
                                         break;
@@ -187,7 +216,7 @@ public class TrackingFlowTask {
                                 TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, correlationContract.get(0));
                                 tkioFlow.setContractCount(contracts.size());
                                 if (tkioFlow != null) {
-                                        tkioFlowList.add(tkioFlow);
+                                    //tkioFlowList.add(tkioFlow);
                                     tkioFlowRepository.save(tkioFlow);
                                 }
                             }
@@ -196,16 +225,17 @@ public class TrackingFlowTask {
                     }
                 }
             }
-                if (tkioFlowList.size() > 0) {
-                    //tkioFlowRepository.save(tkioFlowList);
-                }
-                calculationFlow.setStatus(2);
-                calculationFlowRepository.save(calculationFlow);
+            /*if (tkioFlowList.size() > 0) {
+                tkioFlowRepository.save(tkioFlowList);
+            }*/
+         /*   calculationFlow.setStatus(2);
+            calculationFlowRepository.save(calculationFlow);*/
+            calculationFlowRepository.updateStatus(calculationFlow.getId(), 2);
         } catch (Exception e) {
             logger.error("CalculationFlow:Id::" + calculationFlow.getId() + ":全流量同步失败", e);
-                calculationFlow.setStatus(3);
-                calculationFlowRepository.save(calculationFlow);
-            }
+            /*calculationFlow.setStatus(3);
+            calculationFlowRepository.save(calculationFlow);*/
+            calculationFlowRepository.updateStatus(calculationFlow.getId(), 3);
         }
     }
 
@@ -221,17 +251,22 @@ public class TrackingFlowTask {
     private BigInteger getFlowOnMonths(String email, String yesterday, BigInteger clickNum, List<Contract> contracts) {
         String code = tkioFlowHistoryRepository.findLastRecodeContract(email);
         List<Contract> sortedContract = sortContracts(contracts);
+        BigInteger clickFlow = clickNum;
         boolean skip = true;
+        boolean fistRecode = code == null ? true : false;
         for (int i = 0; i < sortedContract.size(); i++) {
-            if (skip && !sortedContract.get(i).getContractCode().equals(code)) {
-                continue;//跳过已记录的合同
+            if (skip && code != null && !sortedContract.get(i).getContractCode().equals(code)) {
+                continue;//从记录的最后一份合同开始处理
             } else {
                 skip = false;
             }
 
-            TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, sortedContract.get(i), true);
+
+            TkioFlow tkioFlow = getTkioFlow(clickFlow, yesterday, sortedContract.get(i), true);
             Long costFlow = tkioFlow.getCostFlow() == null ? 0L : tkioFlow.getCostFlow();
             TkioFlowHistory flowHistory = new TkioFlowHistory(tkioFlow);
+            flowHistory.setFirstRecode(fistRecode ? 1 : 0);//标记合同的第一次流量记录
+            fistRecode = false;
             flowHistory.setyMonth(DateTime.parse(yesterday).toString("yyyy-MM"));
             flowHistory.setyMonthNext(DateTime.parse(yesterday).plusMonths(1).toString("yyyy-MM"));
             if (costFlow <= 0) {
@@ -247,23 +282,26 @@ public class TrackingFlowTask {
                 if (DateTime.parse(next.getStartDate()).isBefore(DateTime.parse(yesterday))
                         || DateTime.parse(next.getStartDate()).isEqual(DateTime.parse(yesterday))) {
 
+                    // 生效
                     if (DateTime.parse(next.getStartDate()).monthOfYear().get() != new DateTime(next.getCreateTime()).monthOfYear().get()
                             && new DateTime(next.getCreateTime()).dayOfMonth().get() > 10) {
 
                     } else {
-                        //2、生效时 切为非晚录合同  超出流量计入下一合同
-                        clickNum = new BigInteger(costFlow.toString());
+                        //2、生效时 且非晚录合同  超出流量计入下一合同
+                        clickFlow = new BigInteger(costFlow.toString());
+                        fistRecode = true;
                         continue;
                     }
 
                 }
-                //3、未生效或未晚录合同 则关联流量信息到当前合同
+                //3、未生效或晚录合同 则关联流量信息到当前合同
                 if (DateTime.parse(yesterday).monthOfYear().get() == DateTime.parse(next.getStartDate()).monthOfYear().get()) {
-                    //生效当月 超出的流量关联到 下一份合同
-                    clickNum = new BigInteger(costFlow.toString());
+                    //生效当月 超出的流量关联到 该合同
+                    clickFlow = new BigInteger(costFlow.toString());
+                    fistRecode = true;
                     continue;
                 } else {
-                    //生效当月之前 超出的流量计入当前合同
+                    //生效当月之前 超出的流量计入上一合同
                     flowHistory.setNextCode(next.getContractCode());//标记用于那份合同的调整流量
                     tkioFlowHistoryRepository.save(flowHistory);
                 }
@@ -273,7 +311,7 @@ public class TrackingFlowTask {
                 tkioFlowHistoryRepository.save(flowHistory);
             }
         }
-        return clickNum;
+        return clickFlow;
     }
 
     public void task() {
@@ -489,7 +527,7 @@ public class TrackingFlowTask {
         List<Contract> contracts = new ArrayList<>();
 
         Contract c1 = new Contract();
-        c1.setStartDate("2020-10-03");
+        c1.setStartDate("2020-10-02");
         c1.setContractCode("QWWE-20201101-03");
         Contract c2 = new Contract();
         c2.setStartDate("2020-11-02");