소스 검색

添加任务分发逻辑

syh 4 년 전
부모
커밋
338e82979f

+ 10 - 0
jeecg-boot-module-system/src/test/java/org/jeecg/SampleTest.java

@@ -6,6 +6,7 @@ import cn.com.ctop.common.module.entity.UserAllocation;
 import cn.com.ctop.common.module.message.handle.impl.EmailSendMsgHandle;
 import cn.com.ctop.common.module.service.*;
 import cn.com.ctop.common.module.utils.CtopAdConstant;
+import cn.com.ctop.common.module.utils.HttpUtils;
 import cn.com.ctop.kuaishou.modules.batch.service.IKuaiShouHistoryReportTaskService;
 import cn.com.ctop.kuaishou.modules.batch.service.IKuaishouInterfaceService;
 import cn.com.ctop.kuaishou.modules.graphql.service.IKuaishouWebInterfaceService;
@@ -115,6 +116,15 @@ public class SampleTest {
     }
 
     @Test
+    public void testXxlJobDistribute(){
+        String url = "http://127.0.0.1:8090/xxl-job-admin/jobinfo/trigger?id=88&executorParam=";
+        for(int i=0;i<100;i++){
+            String result = HttpUtils.httpPostRequest(url+i,new HashMap<>(),new HashMap<>());
+            System.out.println(result);
+        }
+    }
+
+    @Test
     public void testLoadRulePlanData(){
         Date date = new Date();
         List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();

+ 1 - 8
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/ByteDancePlanRuleDataJob.java

@@ -11,9 +11,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.util.Date;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 @Component
 public class ByteDancePlanRuleDataJob {
@@ -21,7 +18,6 @@ public class ByteDancePlanRuleDataJob {
     private ICtopOauthTokenService tokenService;
     @Autowired
     private IReportService reportService;
-    static ExecutorService executorService = Executors.newFixedThreadPool(10);
 
     @XxlJob("bytedancePlanLoadJob")
     public ReturnT<String> execute(String param) throws Exception {
@@ -30,11 +26,8 @@ public class ByteDancePlanRuleDataJob {
             Long accountId = Long.parseLong(param);
             CtopOauthToken token = tokenService.getTokenByAccountId(accountId);
             reportService.getAdvertiserPlanRuleData(token,date,date,CtopAdConstant.BYTEDANCE_REPORT_TYPE_DAILY);
-        }else{
-            List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();
-            tokens.forEach(token -> executorService.submit(() -> reportService.getAdvertiserPlanRuleData(token,date,date, CtopAdConstant.BYTEDANCE_REPORT_TYPE_DAILY)));
         }
-        XxlJobLogger.log("头条全量计划数据获取完成");
+        XxlJobLogger.log("头条计划数据清洗完成:accountId=>{}",param);
         return ReturnT.SUCCESS;
     }
 }

+ 46 - 0
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/BytedanceRuleDataJobDistribute.java

@@ -0,0 +1,46 @@
+package cn.com.ctop.job.bytedance.handler;
+
+import cn.com.ctop.common.module.entity.CtopOauthToken;
+import cn.com.ctop.common.module.service.ICtopOauthTokenService;
+import cn.com.ctop.common.module.utils.HttpUtils;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import com.xxl.job.core.log.XxlJobLogger;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.List;
+
+@Component
+public class BytedanceRuleDataJobDistribute {
+    @Autowired
+    private ICtopOauthTokenService tokenService;
+    @XxlJob("bytedancePlanRuleDataJobDistribute")
+    public ReturnT<String> bytedancePlanRuleDataJobDistribute(String params) throws Exception {
+        List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();
+        if(null!=tokens&&!tokens.isEmpty()){
+            //TODO 指定相应的任务id
+            String planUrl = "http://139.186.29.76:8090/jobinfo/trigger?id=87&executorParam=";
+            for(CtopOauthToken token:tokens){
+                String planResult = HttpUtils.httpPostRequest(planUrl+token.getAccountId(),new HashMap<>(),new HashMap<>());
+                XxlJobLogger.log("accountId:{};message:{}",token.getAccountId(),planResult);
+            }
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    @XxlJob("bytedanceAccountRuleDataJobDistribute")
+    public ReturnT<String> bytedanceAccountRuleDataJobDistribute(String params) throws Exception {
+        List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();
+        if(null!=tokens&&!tokens.isEmpty()){
+            //TODO
+            String accountUrl = "http://139.186.29.76:8090/jobinfo/trigger?id=87&executorParam=";
+            for(CtopOauthToken token:tokens){
+                String accountResult = HttpUtils.httpPostRequest(accountUrl+token.getAccountId(),new HashMap<>(),new HashMap<>());
+                XxlJobLogger.log("accountId:{};message:{}",token.getAccountId(),accountResult);
+            }
+        }
+        return ReturnT.SUCCESS;
+    }
+}

+ 4 - 13
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/RuleDataAccountCleanJob.java

@@ -3,7 +3,6 @@ package cn.com.ctop.job.bytedance.handler;
 /**
  *  Created by JQ.bi on 2020.11.16
  */
-import cn.com.ctop.common.module.entity.CtopOauthToken;
 import cn.com.ctop.common.module.service.ICtopOauthTokenService;
 import cn.com.ctop.toutiao.modules.report.service.IRuleByteDanceAccountService;
 import com.xxl.job.core.biz.model.ReturnT;
@@ -12,8 +11,6 @@ import com.xxl.job.core.log.XxlJobLogger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-
-import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -31,18 +28,12 @@ public class RuleDataAccountCleanJob {
     @XxlJob("ruleDataAccountCleanJob")
     public ReturnT<String> execute(String param) throws Exception {
         XxlJobLogger.log("规则预警账户维度数据开始清洗");
-        //查询所有头条账户
-        List<CtopOauthToken> tokens = oauthTokenService.selectToutiaoToken();
-        if (null == tokens || tokens.isEmpty()) {
-            XxlJobLogger.log("未获取到可用的token");
+        if (null == param || "".equals(param.trim())) {
+            XxlJobLogger.log("参数异常");
             return ReturnT.FAIL;
         }
-        tokens.forEach(token->executorPool.submit(() -> {
-            ruleByteDanceAccountService.cleanRuleDataAccount(token.getAccountId());
-        }));
-        if(!executorPool.isShutdown()){
-            executorPool.shutdown();
-        }
+        Long accountId = Long.parseLong(param);
+        ruleByteDanceAccountService.cleanRuleDataAccount(accountId);
         XxlJobLogger.log("规则预警账户维度数据清洗结束");
         return ReturnT.SUCCESS;
     }