批量写入selectdb

job-kuaishou/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-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/service/impl/KuaiShouAccountReportServiceImpl.java

@@ -12,21 +12,31 @@ 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 KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportService {
     @Autowired
     private IKuaiShouErrorLogService errorLogService;
     @Autowired
     private KuaiShouAccountReportMapper accountReportMapper;
 
+    private static final int MAX_BATCH_SIZE = 5000;
+    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
+
+    private final List<KuaishouAccountReportHour> advertiserReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
+    private final List<KuaishouAccountReportDaily> advertiserReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
+    private long advertiserReportHourLastFlushTime = System.currentTimeMillis();
+    private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();
+
+
     @Override
     public void queryAccountReportHourly(String accessToken, Long advertiserId, String startDate, String endDate, int page) {
         try {
@@ -73,11 +83,7 @@ public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportS
         }
         if (!Check.isNull(addList)) {
             accountReportMapper.replaceBatcKuaiShouAccountReportHour(addList);
-            try {
-                accountReportMapper.replaceBatcKuaiShouAccountReportHourSD(addList);
-            } catch (Exception e) {
-                log.error("replaceBatcKuaiShouAccountReportHourSD", e);
-            }
+            advertiserReportHourBuffer.addAll(addList);
         }
         if (details.size() >= 2000) {
             queryAccountReportHourly(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -131,11 +137,7 @@ public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportS
         }
         if (!Check.isNull(addList)) {
             accountReportMapper.replaceBatcKuaiShouAccountReportDaily(addList);
-            try {
-                accountReportMapper.replaceBatcKuaiShouAccountReportDailySD(addList);
-            } catch (Exception e) {
-                log.error("replaceBatcKuaiShouAccountReportDailySD", e);
-            }
+            advertiserReportDailyBuffer.addAll(addList);
         }
         if (details.size() >= 2000) {
             queryAccountReportDaily(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -153,4 +155,69 @@ public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportS
     public void updateClAccountStatus() {
         accountReportMapper.updateClAccountStatus();
     }
+
+    @Scheduled(cron = "*/20 * * * * ?") // 每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.replaceBatcKuaiShouAccountReportHourSD(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.replaceBatcKuaiShouAccountReportDailySD(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.replaceBatcKuaiShouAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));                    advertiserReportHourBuffer.clear();
+                    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.replaceBatcKuaiShouAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));                    advertiserReportDailyBuffer.clear();
+                    advertiserReportDailyBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportDaily batch", e);
+                }
+            }
+        }
+    }
 }

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/service/impl/KuaiShouGroupReportServiceImpl.java

@@ -13,15 +13,16 @@ 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.*;
 
 @Slf4j
 @Service
+@EnableScheduling
 public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportService {
     @Autowired
     private KuaiShouGroupReportMapper kuaiShouGroupReportMapper;
@@ -30,6 +31,15 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
     @Autowired
     private IKuaiShouErrorLogService errorLogService;
 
+    private static final int MAX_BATCH_SIZE = 5000;
+    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
+
+    private final List<KuaishouAdgroupReportHourly> reportHourBuffer = Collections.synchronizedList(new ArrayList<>());
+    private final List<KuaishouAdgroupReportDaily> reportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
+    private long reportHourLastFlushTime = System.currentTimeMillis();
+    private long reportDailyLastFlushTime = System.currentTimeMillis();
+
+
 
     @Override
     public void queryGroupReportHourly(String accessToken, Long advertiserId, String startDate, String endDate, int page) {
@@ -80,11 +90,7 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
         }
         if (!Check.isNull(addList)) {
             kuaiShouGroupReportMapper.replaceBatchGroupReportHourly(addList);
-            try {
-                kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(addList);
-            } catch (Exception e) {
-                log.error("replaceBatchGroupReportHourlySD", e);
-            }
+            reportHourBuffer.addAll(addList);
         }
         if (details.size() >= 2000) {
             queryGroupReportHourly(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -153,11 +159,7 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
         }
         if (!Check.isNull(addList)) {
             kuaiShouGroupReportMapper.replaceBatchGroupReportDaily(addList);
-            try {
-                kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(addList);
-            } catch (Exception e) {
-                log.error("replaceBatchGroupReportDailySD", e);
-            }
+            reportDailyBuffer.addAll(addList);
         }
         if (details.size() >= 2000) {
             queryGroupReportDaily(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -166,4 +168,71 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
         }
     }
 
+    @Scheduled(cron = "*/30 * * * * ?") // 每30秒检查一次
+    public void flushIfNeeded() {
+        log.info("开始执行创意报表定时批处理任务....");
+        synchronized (reportHourBuffer) {
+            long now = System.currentTimeMillis();
+            if (reportHourBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - reportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !reportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed creativeReportHour batch size: {}", reportHourBuffer.size());
+                    kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(new ArrayList<>(reportHourBuffer));
+                    reportHourBuffer.clear();
+                    reportHourLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing creativeReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (reportDailyBuffer) {
+            long now = System.currentTimeMillis();
+            if (reportDailyBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - reportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !reportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed creativeReportDaily batch size: {}", reportDailyBuffer.size());
+                    kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(new ArrayList<>(reportDailyBuffer));
+                    reportDailyBuffer.clear();
+                    reportDailyLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing creativeReportDaily batch", e);
+                }
+            }
+        }
+
+
+    }
+
+
+    @PreDestroy
+    public void flushPreDestroy() {
+        log.info("项目重启，开始执行创意报表批处理任务....");
+        synchronized (reportHourBuffer) {
+            if (!reportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed advertiserReportHour batch size: {}", reportHourBuffer.size());
+                    kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(new ArrayList<>(reportHourBuffer));
+                    reportHourBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (reportDailyBuffer) {
+            if (!reportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed creativeReportDaily batch size: {}", reportDailyBuffer.size());
+                    kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(new ArrayList<>(reportDailyBuffer));
+                    reportDailyBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing creativeReportDaily batch", e);
+                }
+            }
+        }
+    }
+
 }

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

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