调整预算走多线程

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/ZtPolicyRunServiceImpl.java

@@ -26,11 +26,15 @@ import java.time.*;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 @Service
 public class ZtPolicyRunServiceImpl implements IZtPolicyRunService {
+    private static final ExecutorService EXTRACTED_EXECUTOR = Executors.newFixedThreadPool(10);
 
     @Resource
     ZtPolicyMapper ztPolicyConfigMapper;
@@ -47,7 +51,10 @@ public class ZtPolicyRunServiceImpl implements IZtPolicyRunService {
 
     @Override
     public void test() {
-        run(30021L);
+        run(60021L);
+//        run(24L);
+//        run(25L);
+//        run(30021L);
     };
 
     @Override
@@ -434,98 +441,129 @@ public class ZtPolicyRunServiceImpl implements IZtPolicyRunService {
 
             }else if (action.getActionType() == 1 || action.getActionType() == 2 || action.getActionType() == 3){//调账号预算
 
+                List<Future<?>> futures = new ArrayList<>();
                 for (String executeTarget : executeTargetList){
-                    actionLog.setTargetId(executeTarget);
+                    ZtActionLog currentActionLog = copyActionLog(actionLog, executeTarget);
                     if (!isOperation(policyId, executeTarget, action.getId(), operateFreqLimit)){
                         //达到频控，不执行
-                        recordActionSkip(actionLog, 2);
+                        recordActionSkip(currentActionLog, 2);
                         continue;
                     }
-                    Integer changeForward =  action.getActionRuleJsonObject().getInteger("changeForward");
-                    BigDecimal changeRate = action.getActionRuleJsonObject().getBigDecimal("changeRate");
-                    BigDecimal changeAmountLimit =  action.getActionRuleJsonObject().getBigDecimal("changeAmountLimit");
-                    JSONObject budgetInfo;
-                    if (action.getActionType() == 1){
-                        budgetInfo = bytedanceBudgetStatusService.getAdvertiserBudget(Long.valueOf(executeTarget));
-                    }else if (action.getActionType() == 2){
-                        String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
-                        budgetInfo = bytedanceBudgetStatusService.getProjectBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget));
-                    }else{
-                        String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
-                        budgetInfo = bytedanceBudgetStatusService.getPromotionBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget));
+                    futures.add(EXTRACTED_EXECUTOR.submit(() -> extracted(policyId, action, executeTarget, policy, operateFreqCircle, currentActionLog)));
+                }
+                for (Future<?> future : futures){
+                    try {
+                        future.get();
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        break;
+                    } catch (Exception e) {
+                        // extracted 内部已做异常处理，这里兜住线程池执行层异常
                     }
-                    if (budgetInfo != null){
-                        String budget_mode =  budgetInfo.getString("budget_mode");
-                        BigDecimal budget = budgetInfo.getBigDecimal("budget");
-                        //计算调整之后的预算
-                        BigDecimal change;
-                        change = budget.multiply(changeRate).divide(new BigDecimal(100), 2, RoundingMode.HALF_UP);
-                        if (changeAmountLimit != null && change.compareTo(changeAmountLimit) > 0){
-                            change = changeAmountLimit;
-                        }
-                        BigDecimal budgetChange;
-                        if (changeForward == 1){
-                            budgetChange = budget.add(change);
-                        }else {
-                            budgetChange = budget.subtract(change);
-                        }
+                }
+                //end for
 
-                        try {
-                            JSONObject operationResult = null;
-                            if (action.getActionType() == 1){
-                                operationResult = bytedanceBudgetStatusService.updateAdvertiserBudget(Long.valueOf(executeTarget), budget_mode, budgetChange);
-                            }else if (action.getActionType() == 2){
-                                String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
-                                operationResult = bytedanceBudgetStatusService.updateProjectBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget), budget_mode, budgetChange);
-                            }else{
-                                String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
-                                operationResult = bytedanceBudgetStatusService.updatePromotionBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget), budget_mode, budgetChange);
-                            }
-                            recordOperation(policyId, executeTarget, action.getId(), operateFreqCircle);
-                            if (operationResult == null){
-                                recordActionError(actionLog, "调整预算失败，调整值："+budgetChange);
-                            }else {
-                                Integer code =  operationResult.getInteger("code");
-                                if (code == null){
-                                    recordActionError(actionLog, "获取调整预算接口结果失败，调整值："+budgetChange);
-                                }else if (code != 0){
-                                    recordActionError(actionLog, "调整预算接口错误, message:"+operationResult.getString("message")+", 调整值:"+budgetChange);
-                                }else {
-                                    recordActionSuccess(actionLog);
-                                }
-                            }
+            }
+        }
 
-                            recordActionSuccess(actionLog);
-                        }catch (Exception e){
-                            String message = e.getMessage();
-                            if (message.length()>500){
-                                message=message.substring(0,500);
-                            }
-                            recordActionError(actionLog, "调整预算出现错误："+message);
-                        }
+    }
+
+    private ZtActionLog copyActionLog(ZtActionLog source, String targetId) {
+        ZtActionLog copied = new ZtActionLog();
+        copied.setId(source.getId());
+        copied.setRunId(source.getRunId());
+        copied.setPolicyId(source.getPolicyId());
+        copied.setTargetType(source.getTargetType());
+        copied.setActionId(source.getActionId());
+        copied.setActionType(source.getActionType());
+        copied.setTargetId(targetId);
+        return copied;
+    }
+
+    private void extracted(Long policyId, ZtPolicyAction action, String executeTarget, ZtPolicy policy, String operateFreqCircle, ZtActionLog actionLog) {
+        Integer changeForward =  action.getActionRuleJsonObject().getInteger("changeForward");
+        BigDecimal changeRate = action.getActionRuleJsonObject().getBigDecimal("changeRate");
+        BigDecimal changeAmountLimit =  action.getActionRuleJsonObject().getBigDecimal("changeAmountLimit");
+        JSONObject budgetInfo;
+        if (action.getActionType() == 1){
+            budgetInfo = bytedanceBudgetStatusService.getAdvertiserBudget(Long.valueOf(executeTarget));
+        }else if (action.getActionType() == 2){
+            String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
+            budgetInfo = bytedanceBudgetStatusService.getProjectBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget));
+        }else{
+            String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
+            budgetInfo = bytedanceBudgetStatusService.getPromotionBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget));
+        }
+        if (budgetInfo != null){
+            String budget_mode =  budgetInfo.getString("budget_mode");
+            BigDecimal budget = budgetInfo.getBigDecimal("budget");
+            //计算调整之后的预算
+            BigDecimal change;
+            change = budget.multiply(changeRate).divide(new BigDecimal(100), 2, RoundingMode.HALF_UP);
+            if (changeAmountLimit != null && change.compareTo(changeAmountLimit) > 0){
+                change = changeAmountLimit;
+            }
+            BigDecimal budgetChange;
+            if (changeForward == 1){
+                budgetChange = budget.add(change);
+            }else {
+                budgetChange = budget.subtract(change);
+            }
+
+            try {
+                JSONObject operationResult = null;
+                if (action.getActionType() == 1){
+                    operationResult = bytedanceBudgetStatusService.updateAdvertiserBudget(Long.valueOf(executeTarget), budget_mode, budgetChange);
+                }else if (action.getActionType() == 2){
+                    String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
+                    operationResult = bytedanceBudgetStatusService.updateProjectBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget), budget_mode, budgetChange);
+                }else{
+                    String advertiserId = policy.getTargetFromAdvertiserId().get(executeTarget);
+                    operationResult = bytedanceBudgetStatusService.updatePromotionBudget(Long.valueOf(advertiserId), Long.valueOf(executeTarget), budget_mode, budgetChange);
+                }
+                recordOperation(policyId, executeTarget, action.getId(), operateFreqCircle);
+                if (operationResult == null){
+                    recordActionError(actionLog, "调整预算失败，调整值："+budgetChange);
+                }else {
+                    Integer code =  operationResult.getInteger("code");
+                    if (code == null){
+                        recordActionError(actionLog, "获取调整预算接口结果失败，调整值："+budgetChange);
+                    }else if (code != 0){
+                        recordActionError(actionLog, "调整预算接口错误, message:"+operationResult.getString("message")+", 调整值:"+budgetChange);
                     }else {
-                        recordActionError(actionLog, "未找到目标的预算信息");
+                        recordActionSuccess(actionLog);
                     }
                 }
-                //end for
 
+                recordActionSuccess(actionLog);
+            }catch (Exception e){
+                String message = e.getMessage();
+                if (message.length()>500){
+                    message=message.substring(0,500);
+                }
+                recordActionError(actionLog, "调整预算出现错误："+message);
             }
+        }else {
+            recordActionError(actionLog, "未找到目标的预算信息");
         }
-
     }
 
+    //记录操作被跳过
     private void recordActionSkip(ZtActionLog log, Integer skipReason){
         recordActionInfo(log, 0, skipReason, 1, null);
     }
 
+    //记录操作失败
     private void recordActionError(ZtActionLog log, String errorMessage){
         recordActionInfo(log, 1, null, 2, errorMessage);
     }
 
+    //记录操作成功
     private void recordActionSuccess(ZtActionLog log){
         recordActionInfo(log, 1, null, 1, null);
     }
 
+    //记录操作信息
     private void recordActionInfo(ZtActionLog log, Integer executed, Integer skipReason, Integer resultStatus, String errorMessage){
         log.setExecuted(executed);
         log.setSkipReason(skipReason);