Przeglądaj źródła

Merge #15 into master from pingkang

selectdb_group_commit

* pingkang: (1 commits)
  selectdb_group_commit

Signed-off-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>
Merged-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>

CR-link: https://codeup.aliyun.com/60c06368dbbed34fe0f0c263/job/hr_job-adq/change/15
huangpingkang 11 miesięcy temu
rodzic
commit
fce49c8cee

+ 2 - 2
job-adq/src/main/java/cn/com/ctop/job/adq/data/mapper/AdqAccountReportMapper.java

@@ -9,11 +9,11 @@ import java.util.List;
 
 public interface AdqAccountReportMapper {
     void addAccountHourList(@Param("adds") List<AdqAdvertiserReportHour> hourList);
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void addAccountHourListSD(@Param("adds") List<AdqAdvertiserReportHour> hourList);
 
     void addAccountDailyList(@Param("adds") List<AdqAdvertiserReportDaily> dailyList);
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void addAccountDailyListSD(@Param("adds") List<AdqAdvertiserReportDaily> dailyList);
 
     void updateAdqAccountStatus();

+ 1 - 1
job-adq/src/main/java/cn/com/ctop/job/adq/data/mapper/AdqAdIncomeMapper.java

@@ -11,7 +11,7 @@ public interface AdqAdIncomeMapper {
 
     void insertAdqAdIncome(List<AdqPublisherAdunitGeneral> reports);
 
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void insertAdqAdIncomeSD(List<AdqPublisherAdunitGeneral> reports);
 
     List<JSONObject> queryAdqIaaMiniApp(@Param("appId")String appId);

+ 2 - 2
job-adq/src/main/java/cn/com/ctop/job/adq/data/mapper/AdqAssetReportMapper.java

@@ -8,10 +8,10 @@ import java.util.List;
 
 public interface AdqAssetReportMapper {
     void insertAdqAssetReportDaily(List<AdqAssetReportDaily> reports);
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void insertAdqAssetReportDailySD(List<AdqAssetReportDaily> reports);
 
     void insertAdqAssetInfo(List<AdqAssetInfo> infos);
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void insertAdqAssetInfoSD(List<AdqAssetInfo> infos);
 }

+ 2 - 2
job-adq/src/main/java/cn/com/ctop/job/adq/data/mapper/AdqCreativeReportMapper.java

@@ -11,13 +11,13 @@ import java.util.List;
 public interface AdqCreativeReportMapper {
     void addsHourList(@Param("adds") List<AdqCreativeReportHour> hourList);
 
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void addsHourListSD(@Param("adds") List<AdqCreativeReportHour> hourList);
 
     List<JSONObject> getAccountList(@Param("accountId") Long accountId);
 
     void addsDaily(@Param("adds") List<AdqCreativeReportDaily> dailies);
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void addsDailySD(@Param("adds") List<AdqCreativeReportDaily> dailies);
 
 }

+ 1 - 1
job-adq/src/main/java/cn/com/ctop/job/adq/data/mapper/AdqFlowDetailsReportDailyMapper.java

@@ -18,7 +18,7 @@ public interface AdqFlowDetailsReportDailyMapper {
     List<AdvertiserAccount> getAdAccount();
 
     void replaceFlowDetailsReportDailies(@Param("list") List<AdqFlowDetailsReportDaily> dailyList);
-    @DS("selectdb")
+    @DS("selectdb_group_commit")
     void replaceFlowDetailsReportDailiesSD(@Param("list") List<AdqFlowDetailsReportDaily> dailyList);
 
 }

+ 8 - 14
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqAccountReportServiceImpl.java

@@ -29,13 +29,13 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
     private AdqErrorLogMapper adqErrorLogMapper;
     @Autowired
     private AdqAccountReportMapper accountReportMapper;
-    private static final int MAX_BATCH_SIZE = 5000;
+    /*private static final int MAX_BATCH_SIZE = 5000;
     private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
 
     private final List<AdqAdvertiserReportHour> advertiserReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
     private final List<AdqAdvertiserReportDaily> advertiserReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
     private long advertiserReportHourLastFlushTime = System.currentTimeMillis();
-    private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();
+    private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();*/
 
     @Override
     public void getAdqAccountHour(Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isLog) {
@@ -97,11 +97,8 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             }
             if (!Check.isNull(hourList)) {
                 accountReportMapper.addAccountHourList(hourList);
-                try {
-                    advertiserReportHourBuffer.addAll(hourList);
-                } catch (Exception e) {
-                    log.error("addAccountHourListSD", e.getMessage());
-                }
+                accountReportMapper.addAccountHourListSD(hourList);
+                //advertiserReportHourBuffer.addAll(hourList);
             }
             if (hourList.size() >= 1000) {
                 getAdqAccountHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -172,11 +169,8 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.addAccountDailyList(dailyList);
-                try {
-                    advertiserReportDailyBuffer.addAll(dailyList);
-                } catch (Exception e) {
-                    log.error("addAccountDailyListSD", e);
-                }
+                accountReportMapper.addAccountDailyListSD(dailyList);
+                //advertiserReportDailyBuffer.addAll(dailyList);
             }
             if (dailyList.size() >= 1000) {
                 getAdqAccountHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -189,7 +183,7 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
     }
 
 
-    @Scheduled(fixedRate = 15000) // 每30秒检查一次
+    /*@Scheduled(fixedRate = 15000) // 每30秒检查一次
     public void flushIfNeeded() {
         log.info("开始执行账户报表定时批处理任务....");
         synchronized (advertiserReportHourBuffer) {
@@ -261,6 +255,6 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
                 }
             }
         }
-    }
+    }*/
 
 }

+ 8 - 17
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqCreativeCreativeReportServiceImpl.java

@@ -34,13 +34,13 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
     private AdqErrorLogMapper adqErrorLogMapper;
 
 
-    private static final int MAX_BATCH_SIZE = 5000;
+    /*private static final int MAX_BATCH_SIZE = 5000;
     private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
 
     private final List<AdqCreativeReportHour> creativeReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
     private final List<AdqCreativeReportDaily> creativeReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
     private long creativeReportHourLastFlushTime = System.currentTimeMillis();
-    private long creativeReportDailyLastFlushTime = System.currentTimeMillis();
+    private long creativeReportDailyLastFlushTime = System.currentTimeMillis();*/
 
 
     @Override
@@ -103,16 +103,10 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             }
             log.info("hourList.size:{}", hourList.size());
             if (!Check.isNull(hourList)) {
-
                 adqReportMapper.addsHourList(hourList);
+                adqReportMapper.addsHourListSD(hourList);
                 log.info("新增数据完成account:id:{}", advertiserId);
-                try {
-                    creativeReportHourBuffer.addAll(hourList);
-
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
-
+                //creativeReportHourBuffer.addAll(hourList);
             }
             if (hourList.size() >= 1000) {
                 getAdqCreativeHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -190,11 +184,8 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             }
             if (!Check.isNull(dailies)) {
                 adqReportMapper.addsDaily(dailies);
-                try {
-                    creativeReportDailyBuffer.addAll(dailies);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+                adqReportMapper.addsDailySD(dailies);
+                //creativeReportDailyBuffer.addAll(dailies);
             }
             if (dailies.size() >= 1000) {
                 getAdqCreativeDaily(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -217,7 +208,7 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
     }
 
 
-    @Scheduled(fixedRate = 10000) // 每30秒检查一次
+    /*@Scheduled(fixedRate = 10000) // 每30秒检查一次
     public void flushIfNeeded() {
         log.info("开始执行创意报表定时批处理任务....");
         synchronized (creativeReportHourBuffer) {
@@ -295,5 +286,5 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
                 }
             }
         }
-    }
+    }*/
 }

+ 6 - 0
job-adq/src/main/resources/application.yml

@@ -93,6 +93,12 @@ spring:
           username: data
           password: hr@2025..
           driver-class-name: com.mysql.jdbc.Driver
+        selectdb_group_commit:
+          ## 注意此selectdb数据源不支持列更新,insert into 对于未指定列会被置为null,类似tidb replace into
+          url: jdbc:mysql://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:9030/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false&useServerPrepStmts=true&useLocalSessionState=true&rewriteBatchedStatements=true&cachePrepStmts=true&prepStmtCacheSqlLimit=99999&prepStmtCacheSize=500&sessionVariables=group_commit=async_mode,enable_nereids_planner=false
+          username: data
+          password: hr@2025..
+          driver-class-name: com.mysql.jdbc.Driver
 
   # redis 配置
   redis: