大表数据清理任务

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/controller/JobController.java

@@ -1,5 +1,6 @@
 package com.tzld.piaoquan.longarticle.controller;
 
+import com.tzld.piaoquan.longarticle.job.DataCleanJob;
 import com.tzld.piaoquan.longarticle.job.NewMatchVideoJob;
 import com.tzld.piaoquan.longarticle.job.RootSourceChannelSyncJob;
 import lombok.extern.slf4j.Slf4j;
@@ -16,6 +17,8 @@ public class JobController {
     NewMatchVideoJob newMatchVideoJob;
     @Autowired
     RootSourceChannelSyncJob rootSourceChannelSyncJob;
+    @Autowired
+    DataCleanJob dataCleanJob;
 
     @GetMapping("/matchCrawlerVideo")
     public void matchCrawlerVideo() {
@@ -41,6 +44,24 @@ public class JobController {
     public void syncTodayRootSourceChannelToRedis(String flowPoolLevel) {
         rootSourceChannelSyncJob.syncTodayRootSourceChannelToRedis(flowPoolLevel);
     }
-}
 
+    @GetMapping("/cleanGroupSendOpenId")
+    public void cleanGroupSendOpenId() {
+        dataCleanJob.cleanGroupSendOpenIdJob(null);
+    }
+
+    @GetMapping("/cleanChangwenDataBaseV2")
+    public void cleanChangwenDataBaseV2() {
+        dataCleanJob.cleanChangwenDataBaseV2Job(null);
+    }
 
+    @GetMapping("/cleanLongArticlesDailyRank")
+    public void cleanLongArticlesDailyRank() {
+        dataCleanJob.cleanLongArticlesDailyRankJob(null);
+    }
+
+    @GetMapping("/cleanI2iRecommend")
+    public void cleanI2iRecommend() {
+        dataCleanJob.cleanI2iRecommendJob(null);
+    }
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/job/DataCleanJob.java

@@ -0,0 +1,103 @@
+package com.tzld.piaoquan.longarticle.job;
+
+import com.tzld.piaoquan.longarticle.utils.LarkRobotUtil;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.temporal.ChronoUnit;
+
+@Slf4j
+@Component
+public class DataCleanJob {
+
+    @Autowired
+    @Qualifier("dataSource")
+    private DataSource dataSource;
+
+    @XxlJob("cleanGroupSendOpenIdJob")
+    public ReturnT<String> cleanGroupSendOpenIdJob(String param) {
+        try {
+            LocalDateTime thresholdDate = getThresholdDateTime();
+            int deletedCount = DayBatchDeleteSupport.deleteByDay(
+                    dataSource,
+                    "long_articles_group_send_open_id",
+                    "create_time",
+                    thresholdDate);
+            log.info("cleanGroupSendOpenIdJob success, thresholdDate={}, deletedCount={}", thresholdDate, deletedCount);
+        } catch (Exception e) {
+            notifyError("long_articles_group_send_open_id");
+            log.error("cleanGroupSendOpenIdJob error:{}", e.getMessage(), e);
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    @XxlJob("cleanChangwenDataBaseV2Job")
+    public ReturnT<String> cleanChangwenDataBaseV2Job(String param) {
+        try {
+            LocalDateTime thresholdDate = getThresholdDateTime();
+            int deletedCount = DayBatchDeleteSupport.deleteByDay(
+                    dataSource,
+                    "changwen_data_base_v2",
+                    "create_time",
+                    thresholdDate);
+            log.info("cleanChangwenDataBaseV2Job success, thresholdDate={}, deletedCount={}", thresholdDate, deletedCount);
+        } catch (Exception e) {
+            notifyError("changwen_data_base_v2");
+            log.error("cleanChangwenDataBaseV2Job error:{}", e.getMessage(), e);
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    @XxlJob("cleanLongArticlesDailyRankJob")
+    public ReturnT<String> cleanLongArticlesDailyRankJob(String param) {
+        try {
+            LocalDate thresholdDate = getThresholdDate();
+            int deletedCount = DayBatchDeleteSupport.deleteByDay(
+                    dataSource,
+                    "long_articles_daily_rank",
+                    "rank_date",
+                    thresholdDate);
+            log.info("cleanLongArticlesDailyRankJob success, thresholdDate={}, deletedCount={}", thresholdDate, deletedCount);
+        } catch (Exception e) {
+            notifyError("long_articles_daily_rank");
+            log.error("cleanLongArticlesDailyRankJob error:{}", e.getMessage(), e);
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    @XxlJob("cleanI2iRecommendJob")
+    public ReturnT<String> cleanI2iRecommendJob(String param) {
+        try {
+            int deletedCount = DayBatchDeleteSupport.deleteByStatus(
+                    dataSource,
+                    "i2i_recommend",
+                    "status",
+                    2);
+            log.info("cleanI2iRecommendJob success, status=2, deletedCount={}", deletedCount);
+        } catch (Exception e) {
+            notifyError("i2i_recommend");
+            log.error("cleanI2iRecommendJob error:{}", e.getMessage(), e);
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    private LocalDateTime getThresholdDateTime() {
+        return LocalDateTime.now(ZoneId.systemDefault()).minus(1, ChronoUnit.MONTHS);
+    }
+
+    private LocalDate getThresholdDate() {
+        return LocalDate.now(ZoneId.systemDefault()).minus(14, ChronoUnit.DAYS);
+    }
+
+    private void notifyError(String tableName) {
+        LarkRobotUtil.sendMessage(tableName + "清理任务异常，请及时查看，<at user_id=\"g6732afb\">王云鹏</at>");
+    }
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/job/DayBatchDeleteSupport.java

@@ -0,0 +1,126 @@
+package com.tzld.piaoquan.longarticle.job;
+
+import lombok.extern.slf4j.Slf4j;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+@Slf4j
+public final class DayBatchDeleteSupport {
+
+    private static final int DELETE_BATCH_SIZE = 10_000;
+
+    private DayBatchDeleteSupport() {
+    }
+
+    public static int deleteByDay(DataSource dataSource, String tableName, String timeColumn, LocalDateTime thresholdDate) throws Exception {
+        List<LocalDate> staleDays = queryStaleDays(dataSource, tableName, timeColumn, thresholdDate);
+        int totalDeletedCount = 0;
+        for (LocalDate staleDay : staleDays) {
+            LocalDateTime dayStart = staleDay.atStartOfDay();
+            LocalDateTime dayEnd = staleDay.plusDays(1).atStartOfDay();
+            LocalDateTime actualEnd = staleDay.equals(thresholdDate.toLocalDate()) ? thresholdDate : dayEnd;
+            if (!actualEnd.isAfter(dayStart)) {
+                continue;
+            }
+            int deletedCount = deleteRange(dataSource, tableName, timeColumn, dayStart, actualEnd);
+            totalDeletedCount += deletedCount;
+            log.info("deleteByDay success, tableName={}, day={}, deletedCount={}", tableName, staleDay, deletedCount);
+        }
+        return totalDeletedCount;
+    }
+
+    public static int deleteByDay(DataSource dataSource, String tableName, String dateColumn, LocalDate thresholdDate) throws Exception {
+        List<LocalDate> staleDays = queryStaleDays(dataSource, tableName, dateColumn, thresholdDate);
+        int totalDeletedCount = 0;
+        for (LocalDate staleDay : staleDays) {
+            int deletedCount = deleteDate(dataSource, tableName, dateColumn, staleDay);
+            totalDeletedCount += deletedCount;
+            log.info("deleteByDay success, tableName={}, day={}, deletedCount={}", tableName, staleDay, deletedCount);
+        }
+        return totalDeletedCount;
+    }
+
+    public static int deleteByStatus(DataSource dataSource, String tableName, String statusColumn, int status) throws Exception {
+        String sql = "delete from " + tableName + " where " + statusColumn + " = ? limit " + DELETE_BATCH_SIZE;
+        int totalDeletedCount = 0;
+        while (true) {
+            int deletedCount;
+            try (Connection connection = dataSource.getConnection();
+                 PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+                preparedStatement.setInt(1, status);
+                deletedCount = preparedStatement.executeUpdate();
+            }
+            totalDeletedCount += deletedCount;
+            if (deletedCount == 0) {
+                return totalDeletedCount;
+            }
+            log.info("deleteByStatus batch success, tableName={}, status={}, deletedCount={}", tableName, status, deletedCount);
+        }
+    }
+
+    private static List<LocalDate> queryStaleDays(DataSource dataSource, String tableName, String timeColumn, LocalDateTime thresholdDate) throws Exception {
+        String sql = "select distinct date(" + timeColumn + ") as cleanup_day " +
+                "from " + tableName + " where " + timeColumn + " < ? order by cleanup_day";
+        List<LocalDate> staleDays = new ArrayList<>();
+        try (Connection connection = dataSource.getConnection();
+             PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+            preparedStatement.setTimestamp(1, Timestamp.valueOf(thresholdDate));
+            try (ResultSet resultSet = preparedStatement.executeQuery()) {
+                while (resultSet.next()) {
+                    java.sql.Date cleanupDay = resultSet.getDate("cleanup_day");
+                    if (cleanupDay != null) {
+                        staleDays.add(cleanupDay.toLocalDate());
+                    }
+                }
+            }
+        }
+        return staleDays;
+    }
+
+    private static List<LocalDate> queryStaleDays(DataSource dataSource, String tableName, String dateColumn, LocalDate thresholdDate) throws Exception {
+        String sql = "select distinct " + dateColumn + " as cleanup_day " +
+                "from " + tableName + " where " + dateColumn + " < ? order by cleanup_day";
+        List<LocalDate> staleDays = new ArrayList<>();
+        try (Connection connection = dataSource.getConnection();
+             PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+            preparedStatement.setDate(1, Date.valueOf(thresholdDate));
+            try (ResultSet resultSet = preparedStatement.executeQuery()) {
+                while (resultSet.next()) {
+                    Date cleanupDay = resultSet.getDate("cleanup_day");
+                    if (cleanupDay != null) {
+                        staleDays.add(cleanupDay.toLocalDate());
+                    }
+                }
+            }
+        }
+        return staleDays;
+    }
+
+    private static int deleteRange(DataSource dataSource, String tableName, String timeColumn, LocalDateTime start, LocalDateTime end) throws Exception {
+        String sql = "delete from " + tableName + " where " + timeColumn + " >= ? and " + timeColumn + " < ?";
+        try (Connection connection = dataSource.getConnection();
+             PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+            preparedStatement.setTimestamp(1, Timestamp.valueOf(start));
+            preparedStatement.setTimestamp(2, Timestamp.valueOf(end));
+            return preparedStatement.executeUpdate();
+        }
+    }
+
+    private static int deleteDate(DataSource dataSource, String tableName, String dateColumn, LocalDate date) throws Exception {
+        String sql = "delete from " + tableName + " where " + dateColumn + " = ?";
+        try (Connection connection = dataSource.getConnection();
+             PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+            preparedStatement.setDate(1, Date.valueOf(date));
+            return preparedStatement.executeUpdate();
+        }
+    }
+}