selecdb批处理

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

@@ -16,6 +16,7 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.PreDestroy;
+import java.text.DecimalFormat;
 import java.util.*;
 
 @Service
@@ -370,16 +371,22 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
         }
     }
 
-    @Scheduled(fixedRate = 20000) // 每30秒检查一次
+    @Scheduled(fixedRate = 12000) // 每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()) {
+            // advertiserReportHourBuffer.size() >= MAX_BATCH_SIZE ||                    (now - advertiserReportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
+            if (!advertiserReportHourBuffer.isEmpty()) {
                 try {
-                    log.info("Flushed advertiserReportHour batch size: {}", advertiserReportHourBuffer.size());
+                    long startTime = System.currentTimeMillis();
                     accountReportMapper.replaceAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));
+                    long endTime = System.currentTimeMillis();
+                    long diffInMilliseconds = endTime - startTime;
+                    double diffInSeconds = diffInMilliseconds / 1000.0;
+                    DecimalFormat df = new DecimalFormat("#.00");
+                    String seconds = df.format(diffInSeconds);
+                    log.info("Flushed advertiserReportHour batch size: {}，执行时间：{}", advertiserReportHourBuffer.size(), seconds);
                     advertiserReportHourBuffer.clear();
                     advertiserReportHourLastFlushTime = now;
                 } catch (Exception e) {
@@ -391,11 +398,17 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
 
         synchronized (advertiserReportDailyBuffer) {
             long now = System.currentTimeMillis();
-            if (advertiserReportDailyBuffer.size() >= MAX_BATCH_SIZE ||
-                    (now - advertiserReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !advertiserReportDailyBuffer.isEmpty()) {
+            // advertiserReportDailyBuffer.size() >= MAX_BATCH_SIZE ||  (now - advertiserReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
+            if (!advertiserReportDailyBuffer.isEmpty()) {
                 try {
-                    log.info("Flushed advertiserReportDaily batch size: {}", advertiserReportDailyBuffer.size());
+                    long startTime = System.currentTimeMillis();
                     accountReportMapper.replaceAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
+                    long endTime = System.currentTimeMillis();
+                    long diffInMilliseconds = endTime - startTime;
+                    double diffInSeconds = diffInMilliseconds / 1000.0;
+                    DecimalFormat df = new DecimalFormat("#.00");
+                    String seconds = df.format(diffInSeconds);
+                    log.info("Flushed advertiserReportDaily batch size: {}，执行时间：{}", advertiserReportDailyBuffer.size(), seconds);
                     advertiserReportDailyBuffer.clear();
                     advertiserReportDailyLastFlushTime = now;
                 } catch (Exception e) {

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

@@ -18,6 +18,7 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.PreDestroy;
+import java.text.DecimalFormat;
 import java.util.*;
 
 @Service
@@ -418,16 +419,23 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
         return promoterReportMapper.getAllList();
     }
 
-    @Scheduled(fixedRate = 30000) // 每30秒检查一次
+    @Scheduled(fixedRate = 11000) // 每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()) {
+            // promoterReportHourBuffer.size() >= MAX_BATCH_SIZE || (now - promoterReportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
+            if ( !promoterReportHourBuffer.isEmpty()) {
                 try {
-                    log.info("Flushed PromoterReportHour batch size: {}", promoterReportHourBuffer.size());
+                    long startTime = System.currentTimeMillis();
                     promoterReportMapper.replacePromoterReportHourSD(new ArrayList<>(promoterReportHourBuffer));
+                    long endTime = System.currentTimeMillis();
+                    long diffInMilliseconds = endTime - startTime;
+                    double diffInSeconds = diffInMilliseconds / 1000.0;
+                    DecimalFormat df = new DecimalFormat("#.00");
+                    String seconds = df.format(diffInSeconds);
+                    log.info("Flushed PromoterReportHour batch size: {}，执行时间：{}", promoterReportHourBuffer.size(),seconds);
+
                     promoterReportHourBuffer.clear();
                     promoterReportHourLastFlushTime = now;
                 } catch (Exception e) {
@@ -439,11 +447,18 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
 
         synchronized (promoterReportDailyBuffer) {
             long now = System.currentTimeMillis();
-            if (promoterReportDailyBuffer.size() >= MAX_BATCH_SIZE ||
-                    (now - promoterReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !promoterReportDailyBuffer.isEmpty()) {
+            // promoterReportDailyBuffer.size() >= MAX_BATCH_SIZE ||  (now - promoterReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
+            if ( !promoterReportDailyBuffer.isEmpty()) {
                 try {
-                    log.info("Flushed PromoterReportDaily batch size: {}", promoterReportDailyBuffer.size());
+                    long startTime = System.currentTimeMillis();
                     promoterReportMapper.replacePromoterReportDailiesSD(new ArrayList<>(promoterReportDailyBuffer));
+                    long endTime = System.currentTimeMillis();
+                    long diffInMilliseconds = endTime - startTime;
+                    double diffInSeconds = diffInMilliseconds / 1000.0;
+                    DecimalFormat df = new DecimalFormat("#.00");
+                    String seconds = df.format(diffInSeconds);
+                    log.info("Flushed PromoterReportDaily batch size: {}，执行时间：{}", promoterReportDailyBuffer.size(),seconds);
+
                     promoterReportDailyBuffer.clear();
                     promoterReportDailyLastFlushTime = now;
                 } catch (Exception e) {