定时清除rank log

community-service/src/main/java/com/tanpu/community/controller/FileController.java

@@ -10,6 +10,7 @@ import com.tanpu.community.api.beans.resp.FileUploadResp;
 import com.tanpu.community.api.enums.OssDirEnum;
 import com.tanpu.community.cache.RedisCache;
 import com.tanpu.community.manager.FileManager;
+import com.tanpu.community.service.RankLogService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cache.caffeine.CaffeineCacheManager;
@@ -28,6 +29,9 @@ public class FileController {
     @Autowired
     private FileManager fileManager;
 
+    @Autowired
+    private RankLogService rankLogService;
+
     @Resource
     private UserHolder userHolder;
 
@@ -47,28 +51,16 @@ public class FileController {
         return CommonResp.success(fileManager.uploadFile(file, OssDirEnum.Theme_Pic, userId));
     }
 
-    @GetMapping("/test")
-    public String test() {
-
-
-//        redisCache.put("11111", JsonUtil.toJson(list), 60);
-//
-//        String v = redisCache.get("11111");
-//        System.out.println(v);
-//        System.out.println(JsonUtil.toJson(JsonUtil.toBean(v, new TypeReference<List<String>>() {
-//        })));
+    @GetMapping("/clearRankLog")
+    public String clearRankLog() {
+        Thread t = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                rankLogService.clearRankLog();
+            }
+        });
 
-//        localCache.getCache("local").put("999", "6666666");
-//        System.out.println((String) localCache.getCache("local").get("999").get());
-//
-//
-//        for (int i = 0; i < 30; i++) {
-//            System.out.println(fileManager.getId("" + i / 2));
-//        }
-//
-//        for (int i = 30; i > 0; i--) {
-//            System.out.println(fileManager.getId("" + i / 2));
-//        }
-        return "";
+        t.start();
+        return "success";
     }
 }

community-service/src/main/java/com/tanpu/community/dao/mapper/community/RankLogMapper.java

@@ -5,6 +5,8 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import org.apache.ibatis.annotations.Param;
 import org.apache.ibatis.annotations.Select;
 
+import java.util.List;
+
 /**
  * <p>
  * 话题排序日志记录’ Mapper 接口
@@ -17,4 +19,7 @@ public interface RankLogMapper extends BaseMapper<RankLogEntity> {
 
     @Select("select max(round) from rank_log where type = #{type}")
     Long selectMaxRound(@Param("type") Integer type);
+
+    @Select("select * from rank_log where type = #{type} order by id asc limit #{batchSize}")
+    List<RankLogEntity> selectByTypeLimit(@Param("type") Integer type, @Param("batchSize") Integer batchSize);
 }
\ No newline at end of file

community-service/src/main/java/com/tanpu/community/manager/ConJobManager.java

 package com.tanpu.community.manager;
 
-import com.tanpu.community.service.RankService;
-import com.tanpu.community.service.RecommendService;
-import com.tanpu.community.service.RedisService;
-import com.tanpu.community.service.VisitLogService;
+import com.tanpu.community.service.*;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
@@ -24,6 +21,9 @@ public class ConJobManager {
     @Autowired
     private RankService rankService;
 
+    @Autowired
+    private RankLogService rankLogService;
+
     @Autowired
     private RecommendService recommendService;
 
@@ -53,4 +53,12 @@ public class ConJobManager {
     public void getThemeNewest() {
         recommendService.refreshNewestThemes();
     }
+
+    /**
+     * 定时把rank_log的日志拿出来，清理数据库
+     */
+    @Scheduled(cron = "0 0 0 ? * 1")
+    public void clearRankLog() {
+        rankLogService.clearRankLog();
+    }
 }

community-service/src/main/java/com/tanpu/community/service/OSSFileService.java

@@ -89,10 +89,21 @@ public class OSSFileService {
 
             return record;
         }else {
-
             throw new BizException("文件格式暂不支持:"+suffix);
         }
+    }
 
+    public void uploadFileNoRecord(byte[] data, String fileName, String fileSuffix, String dirPrefix) {
+        String[] arr = StringUtils.split(fileName, ".");
+        String suffix = arr[arr.length - 1];
+        if (FileTypeEnum.imageTypeSet.contains(suffix)) {
+            //上传
+            String id = uuidGenHelper.getUuidStr();
+            String key = CommunityConstant.OSS_PREFIX_FOLDER + dirPrefix + id + "." + suffix;
+            ossHelper.writeFile(bucketName, key, data, fileSuffix);
+        }else {
+            throw new BizException("文件格式暂不支持:"+suffix);
+        }
     }
 
     private HashMap<String, Integer> getStringIntegerHashMap(byte[] data) {

community-service/src/main/java/com/tanpu/community/service/RankLogService.java

 package com.tanpu.community.service;
 
+import com.alibaba.fastjson.JSON;
 import com.tanpu.common.util.JsonUtil;
 import com.tanpu.community.api.beans.qo.ThemeAnalysDO;
 import com.tanpu.community.api.beans.qo.TopicRankQo;
@@ -7,17 +8,28 @@ import com.tanpu.community.api.enums.RankLogTypeEnum;
 import com.tanpu.community.dao.entity.community.RankLogEntity;
 import com.tanpu.community.dao.mapper.community.RankLogMapper;
 import com.tanpu.community.util.BizUtils;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.time.DateUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
+import java.io.ByteArrayOutputStream;
 import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Date;
 import java.util.List;
+import java.util.stream.Collectors;
 
+@Slf4j
 @Service
 public class RankLogService {
 
+    @Autowired
+    private OSSFileService ossFileService;
 
     @Resource
     private RankLogMapper rankLogMapper;
@@ -79,4 +91,38 @@ public class RankLogService {
         }
     }
 
+
+    // 定时清除ranklog，并上传到oss
+    public void clearRankLog() {
+        LocalDateTime t = LocalDateTime.now().minusDays(7L);
+        log.info("start clearRankLog job before {}", t.format(DateTimeFormatter.BASIC_ISO_DATE));
+
+        for (RankLogTypeEnum type : RankLogTypeEnum.values()) {
+            int idx = 0;
+            while (true) {
+                List<RankLogEntity> logs = rankLogMapper.selectByTypeLimit(type.getCode(), 50);
+                if (logs.isEmpty() || logs.get(0).getCreateTime().isAfter(t)) {
+                    break;
+                }
+
+                try {
+                    String fileName = "ranklog_" + type.getCode() + "_" + t.format(DateTimeFormatter.BASIC_ISO_DATE) + "_" + idx;
+                    ByteArrayOutputStream os = new ByteArrayOutputStream();
+                    IOUtils.writeLines(logs.stream().map(JSON::toJSONString).collect(Collectors.toList()), null, os);
+                    ossFileService.uploadFileNoRecord(os.toByteArray(), fileName, "txt", "");
+
+                    Thread.sleep(1000);
+                } catch (Exception e) {
+                    log.error("error in clearRankLog", e);
+                    throw new RuntimeException(e);
+                }
+
+                // delete
+                rankLogMapper.deleteBatchIds(logs.stream().map(RankLogEntity::getId).collect(Collectors.toList()));
+
+                idx++;
+            }
+        }
+    }
+
 }

community-service/src/main/java/com/tanpu/community/service/RankService.java

@@ -195,7 +195,6 @@ public class RankService {
         //落库
         rankLogService.logTopicRank(rankList, start, TimeUtils.calMillisTillNow(start));
         return;
-
     }
 
     /**