feat: add async throtter

apps/twitter-monitor/src/index.ts

@@ -48,7 +48,9 @@ twitterMonitorUsers$
           concatMap((tweets) =>
             from(tweets).pipe(
               // Process each tweet
+              filter((tweet) => !tweet?.noResults),
               tap((tweet) => {
+                console.log(formatTime(Date.now()), 'tap tweet', user.user_id, tweet);
                 const event: ITwitterEvent = {
                   id: tweet.id,
                   content: tweet.text,

apps/twitter-monitor/src/types/ITwitter.ts

@@ -29,6 +29,7 @@ export interface ITwitter {
   media: any[];
   isConversationControlled: boolean;
   possiblySensitive?: boolean;
+  noResults?: boolean;
 }
 
 export interface Author {

apps/twitter-monitor/src/utils/getUserTweetsByName.ts

@@ -1,5 +1,6 @@
 import { ApifyClient } from 'apify-client';
 import { ITwitter } from '../types/ITwitter';
+import { useAsyncThrottler } from './useAsyncThrotter';
 
 const GET_TWITTER_METHOD_ID = `61RPP7dywgiy0JPD0`;
 const token = process.env.APIFY_TOKEN;
@@ -9,7 +10,7 @@ const client = new ApifyClient({
 });
 
 //通过推特的用户名字 获取推特的信息
-export const getUserTweetsByName = async (name: string): Promise<ITwitter[]> => {
+export const getUserTweetsByName = useAsyncThrottler(async (name: string): Promise<ITwitter[]> => {
   const input = {
     author: name,
     customMapFunction: (object: any) => {
@@ -30,4 +31,4 @@ export const getUserTweetsByName = async (name: string): Promise<ITwitter[]> =>
   // Fetch and print Actor results from the run's dataset (if any)
   const { items } = await client.dataset(run.defaultDatasetId).listItems();
   return items as any;
-};
+}, 30);

apps/twitter-monitor/src/utils/useAsyncThrotter.ts

@@ -0,0 +1,40 @@
+export const useAsyncThrottler = <T, K>(
+  asyncFunction: (params: T) => Promise<K>,
+  concurrencyLimit: number,
+) => {
+  const queue: Array<Function> = []; // 任务队列
+  let activeTasks = 0; // 当前正在执行的任务数量
+  const _next = () => {
+    if (queue.length === 0 || activeTasks >= concurrencyLimit) {
+      return; // 队列为空或达到并发限制，直接返回
+    }
+    // 从队列中取出任务并执行
+    const task = queue.shift();
+    activeTasks++;
+    task?.();
+  };
+  return (innerParams: T): Promise<K> => {
+    return new Promise((resolve, reject) => {
+      // 将任务包装成一个可执行的函数
+      const task = async () => {
+        try {
+          const result = await asyncFunction(innerParams); // 调用异步函数
+          resolve(result); // 任务成功，返回结果
+        } catch (error) {
+          reject(error); // 任务失败，返回错误
+        } finally {
+          activeTasks--; // 任务完成，减少活跃任务数
+          _next(); // 执行下一个任务
+        }
+      };
+
+      // 将任务加入队列
+      queue.push(task);
+
+      // 如果当前活跃任务数未达到限制，立即执行任务
+      if (activeTasks < concurrencyLimit) {
+        _next();
+      }
+    });
+  };
+};

common/changes/@yuants/app-twitter-monitor/2025-02-15-14-40.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@yuants/app-twitter-monitor",
+      "comment": "support async throtter",
+      "type": "patch"
+    }
+  ],
+  "packageName": "@yuants/app-twitter-monitor"
+}