2023年7月13日15:10:51 抓取数据开启线程细节优化

王明元
1 parent 96443f8d
Showing 4 changed files with 226 additions and 5 deletions
src/main/java/cn/fw/freya/FreyaApplication.java
src/main/java/cn/fw/freya/service/CrawlBizService.java
src/main/java/cn/fw/freya/task/DataCaptureTask.java
src/main/java/cn/fw/freya/utils/AutoExpireLocalCache.java
 package cn.fw.freya;
  
+import cn.fw.freya.utils.AutoExpireLocalCache;
 import cn.hutool.core.thread.ThreadFactoryBuilder;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
@@ -46,4 +47,13 @@ public class FreyaApplication {
                 new ThreadFactoryBuilder().setNamePrefix("wmy-thread-pool-").build(),
                 new ThreadPoolExecutor.DiscardPolicy());
     }
+
+
+    /**
+     * 初始化一个自动过期的本地缓存
+     */
+    @Bean
+    public AutoExpireLocalCache initAutoExpireLocalCache() {
+        return new AutoExpireLocalCache(1000);
+    }
 }
@@ -10,11 +10,13 @@ import cn.fw.freya.service.data.AccountService;
 import cn.fw.freya.service.rpc.AccountRpcService;
 import cn.fw.freya.service.rpc.ReportRpcService;
 import cn.fw.freya.utils.AssertUtil;
+import cn.fw.freya.utils.AutoExpireLocalCache;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.event.EventListener;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
  
 import java.io.IOException;
 import java.time.LocalDate;
@@ -39,17 +41,20 @@ public class CrawlBizService {
     private final AccountService accountService;
     private final AccountRpcService accountRpcService;
     private final ReportRpcService reportRpcService;
+    private final AutoExpireLocalCache autoExpireLocalCache;
  
     @Autowired
     public CrawlBizService(final List<CrawlStrategy> crawlStrategyList,
                            final AccountService accountService,
                            final AccountRpcService accountRpcService,
-                           final ReportRpcService reportRpcService) {
+                           final ReportRpcService reportRpcService,
+                           AutoExpireLocalCache autoExpireLocalCache) {
         this.crawlStrategyMap = crawlStrategyList.stream()
                 .collect(Collectors.toMap(CrawlStrategy::getType, Function.identity()));
         this.accountService = accountService;
         this.accountRpcService = accountRpcService;
         this.reportRpcService = reportRpcService;
+        this.autoExpireLocalCache = autoExpireLocalCache;
     }
  
     /**
@@ -102,23 +107,30 @@ public class CrawlBizService {
     public void crawlData(Account account) throws IOException {
         Integer type = account.getType();
         String accountNo = account.getAccountNo();
+        String key = accountNo + "->" + type;
         AccountTypeEnum typeEnum = AccountTypeEnum.getEnumByValue(type);
         CrawlStrategy crawlStrategy = crawlStrategyMap.get(typeEnum);
+        if (StringUtils.hasText(autoExpireLocalCache.get(key)))
+            return;
+        autoExpireLocalCache.put(key, key, 3600 * 1000L);// 1小时自动过期
         // 抓取数据
         log.info("线程: " + Thread.currentThread().getName() + " 开始抓取数据");
         ReportAccountDto accountMsg = crawlStrategy.updateAccountMsg(accountNo);// 更新粉丝数
         if (Objects.isNull(accountMsg)) {
             this.doPushExpireAccount(account);
+            autoExpireLocalCache.removeKey(key);
             return;
         }
         List<VideoPool> allVideoMsg = crawlStrategy.getAllVideoMsg(accountNo);// 获取所有视频信息
         if (Objects.isNull(allVideoMsg)) {
             this.doPushExpireAccount(account);
+            autoExpireLocalCache.removeKey(key);
             return;
         }
         List<LivePool> yesterdayLiveMsg = crawlStrategy.getYesterdayLiveMsg(accountNo);// 获取昨日直播信息
         if (Objects.isNull(yesterdayLiveMsg)) {
             this.doPushExpireAccount(account);
+            autoExpireLocalCache.removeKey(key);
             return;
         }
         // 上报数据
@@ -126,18 +138,22 @@ public class CrawlBizService {
         String format = String.format("上报[%s]平台, 账户号为: %s", AccountTypeEnum.getNameByValue(account.getType()), accountNo);
         if (!reportAccountMsg) {
             log.error(format + "的账户信息失败");
+            autoExpireLocalCache.removeKey(key);
             return;
         }
         boolean reportVideo = this.doReportVideo(account, allVideoMsg);
         if (!reportVideo) {
             log.error(format + "的视频数据失败");
+            autoExpireLocalCache.removeKey(key);
             return;
         }
         boolean reportLive = this.doReportLive(account, yesterdayLiveMsg);
         if (!reportLive) {
             log.error(format + "的直播数据失败");
+            autoExpireLocalCache.removeKey(key);
             return;
         }
+        autoExpireLocalCache.removeKey(key);
         this.afterCrawl(accountService.getById(account.getId()));
     }
  
 package cn.fw.freya.task;
  
-import cn.fw.freya.dao.AccountDao;
 import cn.fw.freya.dao.LivePoolDao;
-import cn.fw.freya.enums.AccountTypeEnum;
 import cn.fw.freya.domain.data.Account;
 import cn.fw.freya.domain.data.pool.LivePool;
+import cn.fw.freya.enums.AccountTypeEnum;
 import cn.fw.freya.service.CrawlBizService;
 import cn.fw.freya.service.crawl.impl.Common;
 import cn.fw.freya.service.crawl.impl.KuaiShouCrawl;
 import cn.fw.freya.service.data.AccountService;
+import cn.fw.freya.utils.AutoExpireLocalCache;
 import cn.fw.freya.utils.DateUtil;
 import cn.fw.freya.utils.PublicUtil;
 import cn.fw.freya.utils.ThreadPoolUtil;
@@ -20,6 +20,7 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
  
 import java.io.IOException;
 import java.util.*;
@@ -44,8 +45,8 @@ public class DataCaptureTask {
     private final Common common;
     private final KuaiShouCrawl kuaiShouCrawl;
     private final LivePoolDao livePoolDao;
-    private final AccountDao accountDao;
     private final ApplicationEventPublisher publisher;
+    private final AutoExpireLocalCache autoExpireLocalCache;
  
     /**
      * 每分钟执行多线程同时抓取数据
@@ -56,12 +57,14 @@ public class DataCaptureTask {
         BlockingQueue<Runnable> queue = threadPoolExecutor.getQueue();// 获取工作队列
         int activeCount = threadPoolExecutor.getActiveCount();// 获取正在执行的线程数
         log.info("当前队列中待执行任务数：" + queue.size() + "\t 正在执行任务的线程数：" + activeCount);
-        if (activeCount >= threadPoolExecutor.getCorePoolSize() && queue.size() > 60) return;
         List<Account> waitCrawlAccount = accountService.findUnCrawlAccount();// 获取今日还未完成抓取数据的账号
         if (CollectionUtils.isEmpty(waitCrawlAccount)) return;
         // 线程池运行多个线程同时抓取多个账号数据
         for (int i = 0; i < waitCrawlAccount.size(); i++) {
             Account currentAccount = waitCrawlAccount.get(i);
+            String key = currentAccount.getAccountNo() + "->" + currentAccount.getType();
+            if (StringUtils.hasText(autoExpireLocalCache.get(key)))
+                continue;
             Account nextAccount;
             if (i < waitCrawlAccount.size() - 1) {
                 nextAccount = waitCrawlAccount.get(i + 1);
+package cn.fw.freya.utils;
+
+import java.util.concurrent.*;
+import java.util.concurrent.locks.LockSupport;
+import java.util.logging.Logger;
+
+/**
+ * @author wmy3969
+ * @version 1.0
+ * @date 2022/8/24 16:06
+ * @Description 带自动过期的本地缓存
+ */
+public class AutoExpireLocalCache {
+    private final DelayQueue<Entity> delayQueue = new DelayQueue<>();// 延迟队列
+    private final ConcurrentHashMap<String, String> cacheMap = new ConcurrentHashMap<>();// 缓存map
+    public static final Logger logger = Logger.getLogger(AutoExpireLocalCache.class.getName());// 日志
+
+    /**
+     * AutoExpireLocalCache对象构造方法, 开启一个线程, 默认100ms一个周期, 监听过期key
+     */
+    public AutoExpireLocalCache() {
+        this(100);
+    }
+
+    /**
+     * AutoExpireLocalCache对象构造方法, 开启一个线程, 指定一个周期(不能小于10ms), 监听过期key
+     *
+     * @param scanCycle 监听过期key, 扫描周期
+     */
+    public AutoExpireLocalCache(long scanCycle) {
+        long cycle = scanCycle <= 10 ? 10 : scanCycle;
+        new Thread(() -> {
+            while (true) {
+                while (!delayQueue.isEmpty()) {
+                    try {
+                        Entity entity = delayQueue.take();
+                        final String waitRemoveKey = entity.getKey();
+                        logger.info("已移除key为: " + waitRemoveKey + ", 队列长度为: " + delayQueue.size() + ", 缓存map元素个数为: " + (cacheMap.size() - 1));
+                        cacheMap.remove(waitRemoveKey);
+                    } catch (InterruptedException e) {
+                        logger.warning("error: " + e.getMessage());
+                        Thread.currentThread().interrupt();
+                    }
+                }
+                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(cycle));
+            }
+        }).start();
+    }
+
+    /**
+     * 缓存map放数据, 默认30s过期
+     *
+     * @param key   待缓存key
+     * @param value 待缓存value
+     */
+    public void put(String key, String value) {
+        this.removeKeyIfAbsent(key);
+        delayQueue.put(new Entity(key));
+        cacheMap.put(key, value);
+    }
+
+    /**
+     * 缓存map放数据
+     *
+     * @param key        待缓存key
+     * @param value      待缓存value
+     * @param saveMillis 保存时间(ms)
+     */
+    public void put(String key, String value, long saveMillis) {
+        this.removeKeyIfAbsent(key);
+        delayQueue.put(new Entity(key, saveMillis));
+        cacheMap.put(key, value);
+    }
+
+    /**
+     * 根据key获取value
+     *
+     * @param key 缓存key
+     * @return 缓存value
+     */
+    public String get(String key) {
+        return cacheMap.get(key);
+    }
+
+    /**
+     * 获取整个缓存map
+     *
+     * @return 整个缓存map
+     */
+    public ConcurrentMap<String, String> getCacheMap() {
+        return cacheMap;
+    }
+
+    /**
+     * 移除缓存key相关内容
+     *
+     * @param key 缓存key
+     */
+    public void removeKey(String key) {
+        if (this.removeKeyIfAbsent(key))
+            cacheMap.remove(key);
+    }
+
+    /**
+     * 获取AutoExpireLocalCache对象中队列长度和map元素个数
+     *
+     * @return 队列长度和map元素个数
+     */
+    public String getSizeMsg() {
+        return "队列长度为: " + delayQueue.size() + ", 缓存map元素个数为: " + cacheMap.size();
+    }
+
+    /**
+     * 销毁AutoExpireLocalCache对象中队列和缓存map
+     */
+    public void clear() {
+        delayQueue.clear();
+        cacheMap.clear();
+    }
+
+    /**
+     * 如果key存在于队列中, 则移除该实体(用作锁续命, 重新指定过期时间)
+     *
+     * @param key 待缓存的key
+     */
+    private boolean removeKeyIfAbsent(String key) {
+        Entity entity = null;
+        if (cacheMap.containsKey(key)) {
+            for (Entity e : delayQueue) {
+                if (e.getKey().equals(key)) {
+                    entity = e;
+                    break;
+                }
+            }
+            if (entity != null)
+                delayQueue.remove(entity);
+            return true;
+        }
+        return false;
+    }
+
+    static class Entity implements Delayed {
+        private final String key;// key
+        private final long createMillis;// 创建时间戳
+        private final long delayMillis;// 延迟时间(ms)
+        private final long expireMillis;// 到期时间戳
+
+        public String getKey() {
+            return key;
+        }
+
+        /**
+         * 不指定key延期时间(默认30s过期)
+         *
+         * @param key 缓存key
+         */
+        public Entity(String key) {
+            this(key, 30000);
+        }
+
+        /**
+         * 指定key延期时间
+         *
+         * @param key         缓存key
+         * @param delayMillis 延期时间(ms)
+         */
+        public Entity(String key, long delayMillis) {
+            this.key = key;
+            this.createMillis = System.currentTimeMillis();
+            this.delayMillis = delayMillis;
+            this.expireMillis = this.createMillis + delayMillis;
+        }
+
+        public long getDelay(TimeUnit unit) {
+            return unit.convert(this.expireMillis - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+        }
+
+        public int compareTo(Delayed delayed) {
+            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - delayed.getDelay(TimeUnit.MILLISECONDS));
+        }
+
+        @Override
+        public String toString() {
+            return "Entity{" +
+                    "key='" + key + '\'' +
+                    ", createMillis=" + createMillis +
+                    ", delayMillis=" + delayMillis +
+                    ", expireMillis=" + expireMillis +
+                    '}';
+        }
+    }
+}