Forráskód Böngészése

adq-批处理数据

yumeng 11 hónapja
szülő
commit
11f5b6dd6f

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

@@ -17,6 +17,7 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 
 
 import javax.annotation.PreDestroy;
 import javax.annotation.PreDestroy;
+import java.text.DecimalFormat;
 import java.time.Instant;
 import java.time.Instant;
 import java.util.*;
 import java.util.*;
 
 
@@ -188,7 +189,7 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
     }
     }
 
 
 
 
-    @Scheduled(fixedRate = 20000) // 每30秒检查一次
+    @Scheduled(fixedRate = 15000) // 每30秒检查一次
     public void flushIfNeeded() {
     public void flushIfNeeded() {
         log.info("开始执行账户报表定时批处理任务....");
         log.info("开始执行账户报表定时批处理任务....");
         synchronized (advertiserReportHourBuffer) {
         synchronized (advertiserReportHourBuffer) {
@@ -196,8 +197,15 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             if (advertiserReportHourBuffer.size() >= MAX_BATCH_SIZE ||
             if (advertiserReportHourBuffer.size() >= MAX_BATCH_SIZE ||
                     (now - advertiserReportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !advertiserReportHourBuffer.isEmpty()) {
                     (now - advertiserReportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !advertiserReportHourBuffer.isEmpty()) {
                 try {
                 try {
-                    log.info("Flushed advertiserReportHour batch size: {}", advertiserReportHourBuffer.size());
+                    long startTime = System.currentTimeMillis();
+
+                    long endTime = System.currentTimeMillis();
+                    long diffInMilliseconds = endTime - startTime;
+                    double diffInSeconds = diffInMilliseconds / 1000.0;
+                    DecimalFormat df = new DecimalFormat("#.00");
+                    String seconds = df.format(diffInSeconds);
                     accountReportMapper.addAccountHourListSD(new ArrayList<>(advertiserReportHourBuffer));
                     accountReportMapper.addAccountHourListSD(new ArrayList<>(advertiserReportHourBuffer));
+                    log.info("Flushed advertiserReportHour batch size: {},执行时间:{}", advertiserReportHourBuffer.size(), seconds);
                     advertiserReportHourBuffer.clear();
                     advertiserReportHourBuffer.clear();
                     advertiserReportHourLastFlushTime = now;
                     advertiserReportHourLastFlushTime = now;
                 } catch (Exception e) {
                 } catch (Exception e) {

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

@@ -19,6 +19,7 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 
 
 import javax.annotation.PreDestroy;
 import javax.annotation.PreDestroy;
+import java.text.DecimalFormat;
 import java.time.Instant;
 import java.time.Instant;
 import java.util.*;
 import java.util.*;
 
 
@@ -216,7 +217,7 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
     }
     }
 
 
 
 
-    @Scheduled(fixedRate = 30000) // 每30秒检查一次
+    @Scheduled(fixedRate = 10000) // 每30秒检查一次
     public void flushIfNeeded() {
     public void flushIfNeeded() {
         log.info("开始执行创意报表定时批处理任务....");
         log.info("开始执行创意报表定时批处理任务....");
         synchronized (creativeReportHourBuffer) {
         synchronized (creativeReportHourBuffer) {
@@ -224,8 +225,14 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             if (creativeReportHourBuffer.size() >= MAX_BATCH_SIZE ||
             if (creativeReportHourBuffer.size() >= MAX_BATCH_SIZE ||
                     (now - creativeReportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !creativeReportHourBuffer.isEmpty()) {
                     (now - creativeReportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !creativeReportHourBuffer.isEmpty()) {
                 try {
                 try {
-                    log.info("Flushed creativeReportHour batch size: {}", creativeReportHourBuffer.size());
+                    long startTime = System.currentTimeMillis();
                     adqReportMapper.addsHourListSD(new ArrayList<>(creativeReportHourBuffer));
                     adqReportMapper.addsHourListSD(new ArrayList<>(creativeReportHourBuffer));
+                    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 creativeReportHour batch size: {},执行时间:{}", creativeReportHourBuffer.size(), seconds);
                     creativeReportHourBuffer.clear();
                     creativeReportHourBuffer.clear();
                     creativeReportHourLastFlushTime = now;
                     creativeReportHourLastFlushTime = now;
                 } catch (Exception e) {
                 } catch (Exception e) {
@@ -240,8 +247,15 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             if (creativeReportDailyBuffer.size() >= MAX_BATCH_SIZE ||
             if (creativeReportDailyBuffer.size() >= MAX_BATCH_SIZE ||
                     (now - creativeReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !creativeReportDailyBuffer.isEmpty()) {
                     (now - creativeReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !creativeReportDailyBuffer.isEmpty()) {
                 try {
                 try {
-                    log.info("Flushed creativeReportDaily batch size: {}", creativeReportDailyBuffer.size());
+                    long startTime = System.currentTimeMillis();
                     adqReportMapper.addsDailySD(new ArrayList<>(creativeReportDailyBuffer));
                     adqReportMapper.addsDailySD(new ArrayList<>(creativeReportDailyBuffer));
+                    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 creativeReportDaily batch size: {},执行时间:{}", creativeReportDailyBuffer.size(),seconds);
+
                     creativeReportDailyBuffer.clear();
                     creativeReportDailyBuffer.clear();
                     creativeReportDailyLastFlushTime = now;
                     creativeReportDailyLastFlushTime = now;
                 } catch (Exception e) {
                 } catch (Exception e) {