2023年11月27日10:16:03 多线程抓取数据加锁逻辑优化

王明元
1 parent 5106cf1f
Showing 8 changed files with 54 additions and 19 deletions
db/freya.mv.db
db/freya.trace.db
src/main/java/cn/fw/freya/FreyaApplication.java
src/main/java/cn/fw/freya/service/CrawlBizService.java
src/main/java/cn/fw/freya/service/data/AccountService.java
src/main/java/cn/fw/freya/service/data/impl/AccountServiceImpl.java
src/main/java/cn/fw/freya/task/DataCaptureTask.java
src/main/java/cn/fw/freya/utils/ThreadPoolUtil.java
@@ -45626,3 +45626,24 @@ SELECT * FROM NS_SIG3 where SIG_MSG like &quot;%noData%&quot; [42122-200]
 	at org.apache.tomcat.util.threads.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:659)
 	at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
 	at java.base/java.lang.Thread.run(Thread.java:829)
+2023-11-27 01:30:10 jdbc[3]: exception
+java.sql.SQLClientInfoException: Client info name 'ApplicationName' not supported.
+	at org.h2.jdbc.JdbcConnection.setClientInfo(JdbcConnection.java:1749)
+	at com.intellij.database.remote.jdbc.impl.RemoteConnectionImpl.setClientInfo(RemoteConnectionImpl.java:466)
+	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
+	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
+	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
+	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
+	at java.rmi/sun.rmi.server.UnicastServerRef.dispatch(UnicastServerRef.java:359)
+	at java.rmi/sun.rmi.transport.Transport$1.run(Transport.java:200)
+	at java.rmi/sun.rmi.transport.Transport$1.run(Transport.java:197)
+	at java.base/java.security.AccessController.doPrivileged(Native Method)
+	at java.rmi/sun.rmi.transport.Transport.serviceCall(Transport.java:196)
+	at java.rmi/sun.rmi.transport.tcp.TCPTransport.handleMessages(TCPTransport.java:562)
+	at java.rmi/sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run0(TCPTransport.java:796)
+	at java.rmi/sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.lambda$run$0(TCPTransport.java:677)
+	at java.base/java.security.AccessController.doPrivileged(Native Method)
+	at java.rmi/sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run(TCPTransport.java:676)
+	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
+	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
+	at java.base/java.lang.Thread.run(Thread.java:834)
@@ -48,7 +48,6 @@ public class FreyaApplication {
                 new ThreadPoolExecutor.DiscardPolicy());
     }
  
-
     /**
      * 初始化一个自动过期的本地缓存
      */
@@ -137,11 +137,9 @@ public class CrawlBizService {
     public void crawlData(Account account) throws IOException {
         Integer type = account.getType();
         String accountNo = account.getAccountNo();
-        String key = account.getCrawlIngLockKey();
         try {
             AccountTypeEnum typeEnum = AccountTypeEnum.getEnumByValue(type);
             CrawlStrategy crawlStrategy = crawlStrategyMap.get(typeEnum);
-            autoExpireLocalCache.put(key, key, TimeUnit.HOURS.toMillis(1));// 1小时自动过期
             /* 抓取数据 */
             log.info("{} 线程开始抓取{}: {} 的数据", Thread.currentThread().getName(), typeEnum.getName(), accountNo);
             ReportAccountDto accountMsg = crawlStrategy.getAccountMsg(accountNo);// 更新粉丝数
@@ -161,7 +159,7 @@ public class CrawlBizService {
             }
             /* 上报数据 */
             boolean reportAccountMsg = this.doReportAccountMsg(account, accountMsg);
-            String format = String.format("上报[%s]平台, 账户号为: %s", AccountTypeEnum.getNameByValue(type), accountNo);
+            String format = String.format("上报[%s]平台, 账户号为: %s", typeEnum.getName(), accountNo);
             if (!reportAccountMsg) {
                 log.error(format + "的账户信息失败");
                 return;
@@ -178,7 +176,7 @@ public class CrawlBizService {
             }
             this.afterCrawl(accountService.getById(account.getId()));
         } finally {
-            autoExpireLocalCache.removeKey(key);
+            autoExpireLocalCache.removeKey(account.getCrawlIngLockKey());
         }
     }
  
@@ -319,7 +317,7 @@ public class CrawlBizService {
         }
         boolean reportLive = this.doReportLive(account, yesterdayLiveMsg);
         if (!reportLive) {
-            accountService.setAccountUndone(accountNo);
+            accountService.setAccountUndone(account);
             log.error(LocalDate.now() + " 上报账户为" + accountNo + "的" +
                     AccountTypeEnum.getNameByValue(account.getType()) + "的直播数据失败");
         }
@@ -106,9 +106,9 @@ public interface AccountService extends IService&lt;Account&gt; {
     /**
      * 将账号标记为未完成
      *
-     * @param accountNo 账号
+     * @param account 账号实体
      */
-    void setAccountUndone(String accountNo);
+    void setAccountUndone(Account account);
  
     /**
      * 获取账号是否已推送过失效待办标识
@@ -234,12 +234,13 @@ public class AccountServiceImpl extends ServiceImpl&lt;AccountDao, Account&gt; impleme
     /**
      * 将账号标记为未完成
      *
-     * @param accountNo 账号
+     * @param account 账号实体
      */
     @Override
-    public void setAccountUndone(String accountNo) {
+    public void setAccountUndone(Account account) {
         this.lambdaUpdate()
-                .eq(Account::getAccountNo, accountNo)
+                .eq(Account::getAccountNo, account.getAccountNo())
+                .eq(Account::getType, account.getType())
                 .set(Account::getDone, false)
                 .update()
         ;
@@ -25,6 +25,7 @@ import org.springframework.util.CollectionUtils;
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
@@ -61,15 +62,23 @@ public class DataCaptureTask {
         List<Account> waitCrawlAccount = accountService.findUnCrawlAccount();// 获取今日还未完成抓取数据的账号
         if (CollectionUtils.isEmpty(waitCrawlAccount)) return;
         for (Account account : waitCrawlAccount) {
-            if (autoExpireLocalCache.containsKey(account.getCrawlIngLockKey()))
+            String key = account.getCrawlIngLockKey();
+            if (autoExpireLocalCache.containsKey(key))
                 continue;
-            threadPoolExecutor.execute(() -> {
-                try {
-                    crawlBizService.crawlData(account);
-                } catch (IOException e) {
-                    log.error("多线程抓取数据发生异常", e);
-                }
-            });
+            try {
+                autoExpireLocalCache.put(key, key, TimeUnit.MINUTES.toMillis(30));// 30分钟自动过期
+                threadPoolExecutor.execute(() -> {
+                    try {
+                        crawlBizService.crawlData(account);
+                    } catch (IOException e) {
+                        log.error("多线程抓取数据发生异常", e);
+                    }
+                });
+            } catch (RejectedExecutionException e) {
+                log.info("多线程抓取: {}({}) 数据被拒绝", account.getAccountNo(),
+                        AccountTypeEnum.getNameByValue(account.getType()));
+                autoExpireLocalCache.removeKey(key);
+            }
         }
     }
  
@@ -67,4 +67,11 @@ public class ThreadPoolUtil {
         getThreadPool().shutdown();
     }
  
+    public static class UserDiscardPolicy implements RejectedExecutionHandler {
+        @Override
+        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+
+        }
+    }
+
 }