فهرست منبع

adq-批处理数据

yumeng 11 ماه پیش
والد
کامیت
e057216967

+ 2 - 2
job-adq/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>
@@ -180,7 +180,7 @@
         <dependency>
             <groupId>redis.clients</groupId>
             <artifactId>jedis</artifactId>
-            <version>2.9.0</version>
+            <version>3.1.0</version>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>

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

@@ -1,8 +1,6 @@
 package cn.com.ctop.job.adq.data.service.impl;
 
-import cn.com.ctop.job.adq.data.entity.AdqAdvertiserReportDaily;
-import cn.com.ctop.job.adq.data.entity.AdqAdvertiserReportHour;
-import cn.com.ctop.job.adq.data.entity.AdqErrorLog;
+import cn.com.ctop.job.adq.data.entity.*;
 import cn.com.ctop.job.adq.data.mapper.AdqAccountReportMapper;
 import cn.com.ctop.job.adq.data.mapper.AdqErrorLogMapper;
 import cn.com.ctop.job.adq.data.service.IAdqAccountReportService;
@@ -14,13 +12,12 @@ 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.Scheduled;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.PreDestroy;
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 @Service
 @Slf4j
@@ -29,7 +26,13 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
     private AdqErrorLogMapper adqErrorLogMapper;
     @Autowired
     private AdqAccountReportMapper accountReportMapper;
+    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();
 
     @Override
     public void getAdqAccountHour(Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isLog) {
@@ -92,7 +95,7 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             if (!Check.isNull(hourList)) {
                 accountReportMapper.addAccountHourList(hourList);
                 try {
-                    accountReportMapper.addAccountHourListSD(hourList);
+                    advertiserReportHourBuffer.addAll(hourList);
                 } catch (Exception e) {
                     log.error("addAccountHourListSD", e.getMessage());
                 }
@@ -167,7 +170,7 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.addAccountDailyList(dailyList);
                 try {
-                    accountReportMapper.addAccountDailyListSD(dailyList);
+                    advertiserReportDailyBuffer.addAll(dailyList);
                 } catch (Exception e) {
                     log.error("addAccountDailyListSD", e);
                 }
@@ -181,4 +184,73 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             e.printStackTrace();
         }
     }
+
+
+    @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.addAccountHourListSD(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.addAccountDailyListSD(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.addAccountHourListSD(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.addAccountDailyListSD(new ArrayList<>(advertiserReportDailyBuffer));
+                    advertiserReportDailyBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportDaily batch", e);
+                }
+            }
+        }
+    }
+
 }

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

@@ -14,8 +14,10 @@ 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.Scheduled;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.PreDestroy;
 import java.time.Instant;
 import java.util.*;
 
@@ -28,6 +30,16 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
     @Autowired
     private AdqErrorLogMapper adqErrorLogMapper;
 
+
+    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();
+
+
     @Override
     public void getAdqCreativeHour(Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isLog) {
         try {
@@ -55,7 +67,6 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             paramsMap.put("fields", fieldsList);
             List<AdqCreativeReportHour> hourList = new ArrayList<>();
             String s = HttpUtils.httpGet("https://api.e.qq.com/v3.0/hourly_reports/get", paramsMap, null);
-            //  System.err.println(s);
             JSONObject jsonObject = JSONObject.parseObject(s);
             if (!Check.isNull(jsonObject)) {
                 Integer code = jsonObject.getInteger("code");
@@ -93,7 +104,8 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
                 adqReportMapper.addsHourList(hourList);
                 log.info("新增数据完成account:id:{}", advertiserId);
                 try {
-                    adqReportMapper.addsHourListSD(hourList);
+                    creativeReportHourBuffer.addAll(hourList);
+
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
@@ -176,7 +188,7 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             if (!Check.isNull(dailies)) {
                 adqReportMapper.addsDaily(dailies);
                 try {
-                    adqReportMapper.addsDailySD(dailies);
+                    creativeReportDailyBuffer.addAll(dailies);
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
@@ -200,4 +212,73 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
     public List<JSONObject> getAccountListByDate(Integer startDate, Integer endDate) {
         return adqErrorLogMapper.getAccountListByDate(startDate, endDate);
     }
+
+
+    @Scheduled(cron = "*/30 * * * * ?") // 每30秒检查一次
+    public void flushIfNeeded() {
+        log.info("开始执行创意报表定时批处理任务....");
+        synchronized (creativeReportHourBuffer) {
+            long now = System.currentTimeMillis();
+            if (creativeReportHourBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - creativeReportHourLastFlushTime) >= MAX_WAIT_TIME_MS && !creativeReportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed creativeReportHour batch size: {}", creativeReportHourBuffer.size());
+                    adqReportMapper.addsHourListSD(new ArrayList<>(creativeReportHourBuffer));
+                    creativeReportHourBuffer.clear();
+                    creativeReportHourLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing creativeReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (creativeReportDailyBuffer) {
+            long now = System.currentTimeMillis();
+            if (creativeReportDailyBuffer.size() >= MAX_BATCH_SIZE ||
+                    (now - creativeReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS && !creativeReportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed creativeReportDaily batch size: {}", creativeReportDailyBuffer.size());
+                    adqReportMapper.addsDailySD(new ArrayList<>(creativeReportDailyBuffer));
+                    creativeReportDailyBuffer.clear();
+                    creativeReportDailyLastFlushTime = now;
+                } catch (Exception e) {
+                    log.error("Error flushing creativeReportDaily batch", e);
+                }
+            }
+        }
+
+
+    }
+
+
+    @PreDestroy
+    public void flushPreDestroy() {
+        log.info("项目重启,开始执行创意报表批处理任务....");
+        synchronized (creativeReportHourBuffer) {
+            if (!creativeReportHourBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed advertiserReportHour batch size: {}", creativeReportHourBuffer.size());
+                    adqReportMapper.addsHourListSD(new ArrayList<>(creativeReportHourBuffer));
+                    creativeReportHourBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing advertiserReportHour batch", e);
+                }
+            }
+        }
+
+
+        synchronized (creativeReportDailyBuffer) {
+            if (!creativeReportDailyBuffer.isEmpty()) {
+                try {
+                    log.info("Flushed creativeReportDaily batch size: {}", creativeReportDailyBuffer.size());
+                    adqReportMapper.addsDailySD(new ArrayList<>(creativeReportDailyBuffer));
+                    creativeReportDailyBuffer.clear();
+                } catch (Exception e) {
+                    log.error("Error flushing creativeReportDaily batch", e);
+                }
+            }
+        }
+    }
+
 }

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

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