selecdb批处理

job-bytedance/pom.xml

@@ -9,7 +9,7 @@
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.1.3.RELEASE</version>
+        <version>2.3.12.RELEASE</version>
         <relativePath/>
     </parent>
     <properties>

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

@@ -1,8 +1,6 @@
 package cn.com.ctop.job.bytedance.data.service.impl;
 
-import cn.com.ctop.job.bytedance.data.entity.BytedanceAdvertiserReportDaily;
-import cn.com.ctop.job.bytedance.data.entity.BytedanceAdvertiserReportHour;
-import cn.com.ctop.job.bytedance.data.entity.BytedanceErrorLog;
+import cn.com.ctop.job.bytedance.data.entity.*;
 import cn.com.ctop.job.bytedance.data.mapper.BytedanceAccountReportMapper;
 import cn.com.ctop.job.bytedance.data.service.IBytedanceAccountReportService;
 import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
@@ -13,19 +11,29 @@ import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import javax.annotation.PreDestroy;
+import java.util.*;
 
 @Service
 @Slf4j
+@EnableScheduling
 public class BytedanceAccountReportServiceImpl implements IBytedanceAccountReportService {
     @Autowired
     private BytedanceAccountReportMapper accountReportMapper;
 
+    private static final int MAX_BATCH_SIZE = 5000;
+    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
+
+    private final List<BytedanceAdvertiserReportHour> advertiserReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
+    private final List<BytedanceAdvertiserReportDaily> advertiserReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
+    private long advertiserReportHourLastFlushTime = System.currentTimeMillis();
+    private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();
+
+
     @Override
     public void getBytedanceRetryAccountHourReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isSend) {
         try {
@@ -79,14 +87,14 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             if (!Check.isNull(hours)) {
                 accountReportMapper.replaceAccountReportHour(hours);
                 try {
-                    accountReportMapper.replaceAccountReportHourSD(hours);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportHourSD",e);
+                    advertiserReportHourBuffer.addAll(hours);
+                } catch (Exception e) {
+                    e.printStackTrace();
                 }
             }
             if (hours.size() >= 100) {
                 try {
-                    Thread.sleep(200L);
+                    Thread.sleep(100L);
                 } catch (InterruptedException e) {
                     throw new RuntimeException(e);
                 }
@@ -154,9 +162,9 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
                 try {
-                    accountReportMapper.replaceAccountReportDailySD(dailyList);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportDailySD",e);
+                    advertiserReportDailyBuffer.addAll(dailyList);
+                } catch (Exception e) {
+                    log.error("replaceAccountReportDailySD", e);
                 }
             }
             if (dailyList.size() >= 100) {
@@ -170,7 +178,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                 log.info("账户ID：{}，日期:{},广告时报（账户粒度）数据请求成功", advertiserId, startDate);
                 accountReportMapper.deleteInfoByErrorId(errorId);
                 if (isSend == 1) {
-                    AlertUtils.sendMessage(advertiserId, startDate, endDate, 1, "hour", 1,"account");
+                    AlertUtils.sendMessage(advertiserId, startDate, endDate, 1, "hour", 1, "account");
                 }
             }
         } catch (Exception e) {
@@ -251,9 +259,9 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             if (!Check.isNull(hours)) {
                 accountReportMapper.replaceAccountReportHour(hours);
                 try {
-                    accountReportMapper.replaceAccountReportHourSD(hours);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportHourSD",e);
+                    advertiserReportHourBuffer.addAll(hours);
+                } catch (Exception e) {
+                    e.printStackTrace();
                 }
             }
             if (hours.size() >= 100) {
@@ -342,9 +350,9 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
                 try {
-                    accountReportMapper.replaceAccountReportDailySD(dailyList);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportDailySD",e);
+                    advertiserReportDailyBuffer.addAll(dailyList);
+                } catch (Exception e) {
+                    log.error("replaceAccountReportDailySD", e);
                 }
             }
             if (dailyList.size() >= 100) {
@@ -361,4 +369,72 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             e.printStackTrace();
         }
     }
+
+    @Scheduled(cron = "*/30 * * * * ?") // 每30秒检查一次
+    public void flushIfNeeded() {
+        log.info("开始执行账户报表定时批处理任务....");
+        synchronized (advertiserReportHourBuffer) {
+            long now = System.currentTimeMillis();
+            if (advertiserReportHourBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - advertiserReportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !advertiserReportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed advertiserReportHour batch size: {}", advertiserReportHourBuffer.size());
+                    accountReportMapper.replaceAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));
+                    advertiserReportHourBuffer.clear();
+                    advertiserReportHourLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (advertiserReportDailyBuffer) {
+            long now = System.currentTimeMillis();
+            if (advertiserReportDailyBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - advertiserReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !advertiserReportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed advertiserReportDaily batch size: {}", advertiserReportDailyBuffer.size());
+                    accountReportMapper.replaceAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
+                    advertiserReportDailyBuffer.clear();
+                    advertiserReportDailyLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportDaily batch", e);
+                }
+            }
+        }
+
+
+    }
+
+
+    @PreDestroy
+    public void flushPreDestroy() {
+        log.info("项目重启，开始执行广告报表批处理任务....");
+        synchronized (advertiserReportHourBuffer) {
+            if (!advertiserReportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed advertiserReportHour batch size: {}", advertiserReportHourBuffer.size());
+                    accountReportMapper.replaceAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));
+                    advertiserReportHourBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (advertiserReportDailyBuffer) {
+            if (!advertiserReportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed advertiserReportDaily batch size: {}", advertiserReportDailyBuffer.size());
+                    accountReportMapper.replaceAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
+                    advertiserReportDailyBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportDaily batch", e);
+                }
+            }
+        }
+    }
+
 }

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

@@ -13,18 +13,26 @@ import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import javax.annotation.PreDestroy;
+import java.util.*;
 
 @Service
 @Slf4j
+@EnableScheduling
 public class BytedancePromoterReportServiceImpl implements IBytedancePromoterReportService {
     @Autowired
     private BytedancePromoterReportMapper promoterReportMapper;
+    private static final int MAX_BATCH_SIZE = 10000;
+    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
+    private final List<BytedancePromoterReportHour> promoterReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
+    private final List<BytedancePromoterReportDaily> promoterReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
+    private long promoterReportHourLastFlushTime = System.currentTimeMillis();
+    private long promoterReportDailyLastFlushTime = System.currentTimeMillis();
+
 
     @Override
     public List<JSONObject> getAccountList() {
@@ -92,7 +100,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
                 try {
-                    promoterReportMapper.replacePromoterReportHourSD(hours);
+                    promoterReportHourBuffer.addAll(hours);
                 } catch (Exception e) {
                     log.error("", e);
                 }
@@ -176,7 +184,8 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             if (!Check.isNull(promoterReportDailies)) {
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
                 try {
-                    promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
+                    promoterReportDailyBuffer.addAll(promoterReportDailies);
+
                 } catch (Exception e) {
                     log.error("", e);
                 }
@@ -282,7 +291,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
                 try {
-                    promoterReportMapper.replacePromoterReportHourSD(hours);
+                    promoterReportHourBuffer.addAll(hours);
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
@@ -320,10 +329,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             Map<String, String> headers = new HashMap<>();
             headers.put("Access-Token", accessToken);
             headers.put("Content-Type", "application/json");
-            long urlStartTime = System.currentTimeMillis();
             String s = HttpUtils.httpGet(url, paramsMap, headers);
-            long urlEndTime = System.currentTimeMillis();
-            log.info("urlTime:{}", urlEndTime - urlStartTime);
             JSONObject result = JSONObject.parseObject(s);
             if (Check.isNull(result)) {
                 log.info("账户id:{}，广告日报（最细粒度）请求结果返回为空", advertiserId);
@@ -380,26 +386,15 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                 return;
             }
             if (!Check.isNull(promoterReportDailies)) {
-                long tidbStartTime = System.currentTimeMillis();
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
-                long tidbEndTime = System.currentTimeMillis();
-                log.info("tidb入库时间:{}", tidbEndTime - tidbStartTime);
                 try {
-                    long selectDbStartTime = System.currentTimeMillis();
-                    promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
-                    long selectDbEnsTime = System.currentTimeMillis();
-                    log.info("selectDb入库时间:{}", selectDbEnsTime - selectDbStartTime);
+                    promoterReportDailyBuffer.addAll(promoterReportDailies);
                 } catch (Exception e) {
                     log.error("", e);
                 }
                 log.info("账户ID：{}，广告日报（最细粒度）数据请求成功", advertiserId);
             }
             if (promoterReportDailies.size() >= 100) {
-          /*      try {
-                    Thread.sleep(200L);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }*/
                 getBytedancePromoterDaily(advertiserId, accessToken, startDate, endDate, page + 1, 1);
             } else {
                 log.info("账户ID：{}，开始日期:{},结束日期:{},广告日报（最细粒度）数据请求成功", advertiserId, startDate, endDate);
@@ -423,4 +418,72 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
         return promoterReportMapper.getAllList();
     }
 
+    @Scheduled(cron = "*/30 * * * * ?") // 每30秒检查一次
+    public void flushIfNeeded() {
+        log.info("开始执行广告报表定时批处理任务....");
+        synchronized (promoterReportHourBuffer) {
+            long now = System.currentTimeMillis();
+            if (promoterReportHourBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - promoterReportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !promoterReportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed PromoterReportHour batch size: {}", promoterReportHourBuffer.size());
+                    promoterReportMapper.replacePromoterReportHourSD(new ArrayList<>(promoterReportHourBuffer));
+                    promoterReportHourBuffer.clear();
+                    promoterReportHourLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing PromoterReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (promoterReportDailyBuffer) {
+            long now = System.currentTimeMillis();
+            if (promoterReportDailyBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - promoterReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !promoterReportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed PromoterReportDaily batch size: {}", promoterReportDailyBuffer.size());
+                    promoterReportMapper.replacePromoterReportDailiesSD(new ArrayList<>(promoterReportDailyBuffer));
+                    promoterReportDailyBuffer.clear();
+                    promoterReportDailyLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing PromoterReportDaily batch", e);
+                }
+            }
+        }
+
+
+    }
+
+
+    @PreDestroy
+    public void flushPreDestroy() {
+        log.info("项目重启，开始执行广告报表批处理任务....");
+        synchronized (promoterReportHourBuffer) {
+            if (!promoterReportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed PromoterReportHour batch size: {}", promoterReportHourBuffer.size());
+                    promoterReportMapper.replacePromoterReportHourSD(new ArrayList<>(promoterReportHourBuffer));
+                    promoterReportHourBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing PromoterReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (promoterReportDailyBuffer) {
+            if (!promoterReportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed promoterReportDaily batch size: {}", promoterReportDailyBuffer.size());
+                    promoterReportMapper.replacePromoterReportDailiesSD(new ArrayList<>(promoterReportDailyBuffer));
+                    promoterReportDailyBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing promoterReportDaily batch", e);
+                }
+            }
+        }
+    }
+
+
 }

job-bytedance/src/main/resources/application.yml

@@ -1,6 +1,7 @@
 server:
   port: 8085
   ip: 127.0.0.1
+  shutdown: graceful   #IMMEDIATE立即停机（默认值）、GRACEFUL
   servlet:
     context-path: /
     compression: