feature(*): 新增历史数据清理定时任务

- 新增历史数据清理定时任务

feature(*): 新增历史数据清理定时任务
- 新增历史数据清理定时任务
张志伟
1 parent 4764c95d
Showing 5 changed files with 116 additions and 2 deletions
fw-hestia-common/pom.xml
fw-hestia-common/src/main/java/cn/fw/hestia/common/utils/ThreadPoolUtil.java
fw-hestia-server/src/main/java/cn/fw/hestia/server/task/SendMessageTask.java
fw-hestia-service/src/main/java/cn/fw/hestia/service/buz/MessageCenterBizService.java
pom.xml
@@ -16,6 +16,10 @@
  
     <dependencies>
         <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+        </dependency>
+        <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
         </dependency>
+package cn.fw.hestia.common.utils;
+
+import cn.hutool.core.thread.ExecutorBuilder;
+import cn.hutool.core.thread.NamedThreadFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * ThreadPoolUtil
+ *
+ * @author : kurisu
+ * @version : 2.0
+ * @className : ThreadPoolUtil
+ * @description : ThreadPoolUtil
+ * @date : 2023-04-19 17:31
+ */
+public class ThreadPoolUtil {
+    private static volatile ThreadPoolUtil INSTANCE;
+    private final ExecutorService executor;
+
+    private ThreadPoolUtil() {
+        this.executor = ExecutorBuilder.create()
+                .setCorePoolSize(20)
+                .setMaxPoolSize(200)
+                .setThreadFactory(new NamedThreadFactory("hestia-custom-", false))
+                .setAllowCoreThreadTimeOut(true)
+                .setWorkQueue(new LinkedBlockingQueue<>())
+                .build();
+    }
+
+    public static ThreadPoolUtil getInstance() {
+        if (INSTANCE == null) {
+            synchronized (ThreadPoolUtil.class) {
+                if (INSTANCE == null) {
+                    INSTANCE = new ThreadPoolUtil();
+                }
+            }
+        }
+        return INSTANCE;
+    }
+
+    public ExecutorService getExecutor() {
+        return executor;
+    }
+}
 package cn.fw.hestia.server.task;
  
+import cn.fw.hestia.common.utils.ThreadPoolUtil;
 import cn.fw.hestia.domain.db.MessageHistory;
 import cn.fw.hestia.domain.enums.MessageStateEnum;
 import cn.fw.hestia.service.buz.MessageCenterBizService;
 import cn.fw.hestia.service.data.MessageHistoryService;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import lombok.Getter;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.core.BoundSetOperations;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
  
+import java.time.LocalDateTime;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -28,12 +34,19 @@ import static cn.fw.hestia.common.constant.MessageConstant.MAX_FREQUENCY;
 public class SendMessageTask {
     private final MessageCenterBizService messageCenterBizService;
     private final MessageHistoryService messageHistoryService;
+    private final StringRedisTemplate stringRedisTemplate;
+
+    @Getter
+    @Value("${spring.cache.custom.global-prefix}:data:clear")
+    private String dataClearKey;
  
     @Autowired
     public SendMessageTask(final MessageCenterBizService messageCenterBizService,
-                           final MessageHistoryService messageHistoryService) {
+                           final MessageHistoryService messageHistoryService,
+                           final StringRedisTemplate stringRedisTemplate) {
         this.messageCenterBizService = messageCenterBizService;
         this.messageHistoryService = messageHistoryService;
+        this.stringRedisTemplate = stringRedisTemplate;
     }
  
     /**
@@ -55,4 +68,35 @@ public class SendMessageTask {
             CompletableFuture.runAsync(() -> messageCenterBizService.sendMessage(history));
         }
     }
+
+    /**
+     * 缓存数据
+     */
+    @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 10)
+    public void cacheOldData() {
+        String key = getDataClearKey();
+        final BoundSetOperations<String, String> setOps = stringRedisTemplate.boundSetOps(key);
+        List<MessageHistory> list = messageHistoryService.list(Wrappers.<MessageHistory>lambdaQuery()
+                .lt(MessageHistory::getCreateTime, LocalDateTime.now().minusMonths(6L))
+                .last("limit 500")
+        );
+        if (CollectionUtils.isEmpty(list)) {
+            return;
+        }
+        list.stream().map(MessageHistory::getId).forEach(id -> setOps.add(id.toString()));
+    }
+
+    /**
+     * 清理数据
+     */
+    @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 15)
+    public void clearOldData() {
+        String key = getDataClearKey();
+        final BoundSetOperations<String, String> setOps = stringRedisTemplate.boundSetOps(key);
+        String idStr;
+        while ((idStr = setOps.pop()) != null) {
+            final Long id = Long.valueOf(idStr);
+            CompletableFuture.runAsync(() -> messageCenterBizService.clearHistory(id), ThreadPoolUtil.getInstance().getExecutor());
+        }
+    }
 }
@@ -131,6 +131,20 @@ public class MessageCenterBizService {
         messageHistoryService.updateById(history);
     }
  
+
+    /**
+     * 清理不使用的数据
+     *
+     * @param id
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void clearHistory(Long id) {
+        sendLogService.remove(Wrappers.<SendLog>lambdaQuery()
+                .eq(SendLog::getMessageId, id)
+        );
+        messageHistoryService.removeById(id);
+    }
+
     /**
      * 撤回消息
      *
@@ -12,7 +12,7 @@
     <parent>
         <groupId>cn.fw</groupId>
         <artifactId>fw-common-dependencies</artifactId>
-        <version>3.3.2</version>
+        <version>3.3.3</version>
     </parent>
  
     <modules>
@@ -38,6 +38,7 @@
         <fastjson>1.2.51</fastjson>
         <fw-passport-sdk.version>2.2.0</fw-passport-sdk.version>
         <yitter.idgenerator>1.0.6</yitter.idgenerator>
+        <hutool.version>5.2.5</hutool.version>
     </properties>
  
     <dependencyManagement>
@@ -121,6 +122,11 @@
                 <artifactId>yitter-idgenerator</artifactId>
                 <version>${yitter.idgenerator}</version>
             </dependency>
+            <dependency>
+                <groupId>cn.hutool</groupId>
+                <artifactId>hutool-all</artifactId>
+                <version>${hutool.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>