update:手写队列

hjdhnx · hjdhnx · commit 431ea1e2e7c3 · 2024-12-09T02:59:14.000+08:00
diff --git a/drop_code/batchFetch.js b/drop_code/batchFetch.js
@@ -0,0 +1,33 @@
+import PQueue from 'p-queue';
+import axios from "../libs_drpy/axios.min";
+
+// p-queue 的实现，不怎么兼容海阔的旧版
+globalThis.batchFetch = async (items, maxWorkers = 5, timeoutConfig = 5000) => {
+    const queue = new PQueue({concurrency: maxWorkers});
+
+    // 获取全局 timeout 设置
+    const timeout = timeoutConfig;
+
+    // 遍历 items 并生成任务队列
+    const promises = items.map((item) => {
+        return queue.add(async () => {
+            try {
+                const response = await axios(
+                    Object.assign({}, item?.options, {
+                        url: item.url,
+                        method: item?.options?.method || 'GET',
+                        timeout: item?.options?.timeout || timeout,
+                        responseType: 'text',
+                    }),
+                );
+                return response.data;
+            } catch (error) {
+                console.log(`[batchFetch][error] ${item.url}: ${error}`);
+                return null;
+            }
+        });
+    });
+
+    // 执行所有任务
+    return Promise.all(promises);
+};
diff --git a/js/PTT[优].js b/js/PTT[优].js
@@ -101,6 +101,7 @@ var rule = {
         lists.push(list1);
         // log(v_tab_urls);
         if (v_tab_urls.length > 1) {
+            let t1 = (new Date()).getTime();
             let reqUrls = v_tab_urls.slice(1).map(it => {
                 return {
                     url: it,
@@ -119,6 +120,8 @@ var rule = {
                     lists.push([]);
                 }
             });
+            let t2 = (new Date()).getTime();
+            log(`批量请求二级 ${input} 耗时${t2 - t1}毫秒:`);
         }
         let playUrls = lists.map(it => it.join('#'));
         VOD.vod_play_url = playUrls.join('$$$');
diff --git a/libs_drpy/drpyInject.js b/libs_drpy/drpyInject.js
@@ -1,7 +1,9 @@
 // import axios, {toFormData} from 'axios';
 import axios, {toFormData} from './axios.min.js';
 
-import PQueue from 'p-queue';
+// import PQueue from 'p-queue';
+// import Queue from 'queue';
+import Queue from './queue.js';
 import crypto from 'crypto';
 import https from 'https';
 import fs from 'node:fs';
@@ -138,15 +140,84 @@ async function request(url, opt = {}) {
     return {headers: {}, content: ''};
 }
 
+// globalThis.batchFetch = async (items, maxWorkers = 5, timeoutConfig = 5000) => {
+//     const queue = new PQueue({concurrency: maxWorkers});
+//
+//     // 获取全局 timeout 设置
+//     const timeout = timeoutConfig;
+//
+//     // 遍历 items 并生成任务队列
+//     const promises = items.map((item) => {
+//         return queue.add(async () => {
+//             try {
+//                 const response = await axios(
+//                     Object.assign({}, item?.options, {
+//                         url: item.url,
+//                         method: item?.options?.method || 'GET',
+//                         timeout: item?.options?.timeout || timeout,
+//                         responseType: 'text',
+//                     }),
+//                 );
+//                 return response.data;
+//             } catch (error) {
+//                 console.log(`[batchFetch][error] ${item.url}: ${error}`);
+//                 return null;
+//             }
+//         });
+//     });
+//
+//     // 执行所有任务
+//     return Promise.all(promises);
+// };
+
+// globalThis.batchFetch = async (items, maxWorkers = 5, timeoutConfig = 5000) => {
+//     const queue = new Queue({concurrency: maxWorkers, autostart: true});
+//
+//     // 获取全局 timeout 设置
+//     const timeout = timeoutConfig;
+//
+//     const results = [];
+//     const promises = [];
+//
+//     items.forEach((item, index) => {
+//         promises.push(
+//             new Promise((resolve) => {
+//                 queue.push(async () => {
+//                     try {
+//                         const response = await axios(
+//                             Object.assign({}, item?.options, {
+//                                 url: item.url,
+//                                 method: item?.options?.method || 'GET',
+//                                 timeout: item?.options?.timeout || timeout,
+//                                 responseType: 'text',
+//                             }),
+//                         );
+//                         results[index] = response.data;
+//                         resolve();
+//                     } catch (error) {
+//                         console.log(`[batchFetch][error] ${item.url}: ${error}`);
+//                         results[index] = null;
+//                         resolve();
+//                     }
+//                 });
+//             }),
+//         );
+//     });
+//
+//     await Promise.all(promises);
+//     return results;
+// };
+
 globalThis.batchFetch = async (items, maxWorkers = 5, timeoutConfig = 5000) => {
-    const queue = new PQueue({concurrency: maxWorkers});
+    const queue = new Queue(maxWorkers);
 
     // 获取全局 timeout 设置
     const timeout = timeoutConfig;
 
-    // 遍历 items 并生成任务队列
-    const promises = items.map((item) => {
-        return queue.add(async () => {
+    const results = [];
+
+    items.forEach((item, index) => {
+        queue.add(async () => {
             try {
                 const response = await axios(
                     Object.assign({}, item?.options, {
@@ -156,16 +227,16 @@ globalThis.batchFetch = async (items, maxWorkers = 5, timeoutConfig = 5000) => {
                         responseType: 'text',
                     }),
                 );
-                return response.data;
+                results[index] = response.data;
             } catch (error) {
                 console.log(`[batchFetch][error] ${item.url}: ${error}`);
-                return null;
+                results[index] = null;
             }
         });
     });
 
-    // 执行所有任务
-    return Promise.all(promises);
+    await queue.onIdle();
+    return results;
 };
 
 function base64EncodeBuf(buff, urlsafe = false) {
diff --git a/libs_drpy/queue.js b/libs_drpy/queue.js
@@ -0,0 +1,44 @@
+class Queue {
+    constructor(concurrency = 1) {
+        this.concurrency = concurrency;
+        this.queue = [];
+        this.activeCount = 0;
+    }
+
+    async runTask(task) {
+        this.activeCount++;
+        try {
+            await task();
+        } catch (err) {
+            console.error('Task failed:', err);
+        } finally {
+            this.activeCount--;
+            this.next();
+        }
+    }
+
+    next() {
+        if (this.queue.length > 0 && this.activeCount < this.concurrency) {
+            const nextTask = this.queue.shift();
+            this.runTask(nextTask);
+        }
+    }
+
+    add(task) {
+        this.queue.push(task);
+        this.next();
+    }
+
+    onIdle() {
+        return new Promise((resolve) => {
+            const interval = setInterval(() => {
+                if (this.queue.length === 0 && this.activeCount === 0) {
+                    clearInterval(interval);
+                    resolve();
+                }
+            }, 10);
+        });
+    }
+}
+
+export default Queue;
diff --git a/package.json b/package.json
@@ -16,7 +16,6 @@
     "fastify": "^4.15.0",
     "lodash": "^4.17.21",
     "mime-types": "^2.1.35",
-    "p-queue": "^8.0.1",
     "qs": "^6.13.1",
     "rotating-file-stream": "^3.2.5",
     "tunnel": "^0.0.6"
diff --git a/yarn.lock b/yarn.lock
@@ -353,11 +353,6 @@ escape-html@~1.0.3:
   resolved "https://registry.npmmirror.com/escape-html/-/escape-html-1.0.3.tgz#0258eae4d3d0c0974de1c169188ef0051d1d1988"
   integrity sha512-NiSupZ4OeuGwr68lGIeym/ksIZMJodUGOSCZ/FSnTxcrekbvqrgdUxlJOMpijaKZVjAJrWrGs/6Jy8OMuyj9ow==
 
-eventemitter3@^5.0.1:
-  version "5.0.1"
-  resolved "https://registry.npmmirror.com/eventemitter3/-/eventemitter3-5.0.1.tgz#53f5ffd0a492ac800721bb42c66b841de96423c4"
-  integrity sha512-GWkBvjiSZK87ELrYOSESUYeVIc9mvLLf/nXalMOS5dYrgZq9o5OVkbZAVM06CVxYsCwH9BDZFPlQTlPA1j4ahA==
-
 fast-content-type-parse@^1.1.0:
   version "1.1.0"
   resolved "https://registry.npmmirror.com/fast-content-type-parse/-/fast-content-type-parse-1.1.0.tgz#4087162bf5af3294d4726ff29b334f72e3a1092c"
@@ -653,19 +648,6 @@ on-exit-leak-free@^2.1.0:
   resolved "https://registry.npmmirror.com/on-exit-leak-free/-/on-exit-leak-free-2.1.2.tgz#fed195c9ebddb7d9e4c3842f93f281ac8dadd3b8"
   integrity sha512-0eJJY6hXLGf1udHwfNftBqH+g73EU4B504nZeKpz1sYRKafAghwxEJunB2O7rDZkL4PGfsMVnTXZ2EjibbqcsA==
 
-p-queue@^8.0.1:
-  version "8.0.1"
-  resolved "https://registry.npmmirror.com/p-queue/-/p-queue-8.0.1.tgz#718b7f83836922ef213ddec263ff4223ce70bef8"
-  integrity sha512-NXzu9aQJTAzbBqOt2hwsR63ea7yvxJc0PwN/zobNAudYfb1B7R08SzB4TsLeSbUCuG467NhnoT0oO6w1qRO+BA==
-  dependencies:
-    eventemitter3 "^5.0.1"
-    p-timeout "^6.1.2"
-
-p-timeout@^6.1.2:
-  version "6.1.3"
-  resolved "https://registry.npmmirror.com/p-timeout/-/p-timeout-6.1.3.tgz#9635160c4e10c7b4c3db45b7d5d26f911d9fd853"
-  integrity sha512-UJUyfKbwvr/uZSV6btANfb+0t/mOhKV/KXcCUTp8FcQI+v/0d+wXqH4htrW0E4rR6WiEO/EPvUFiV9D5OI4vlw==
-
 package-json-from-dist@^1.0.0:
   version "1.0.1"
   resolved "https://registry.npmmirror.com/package-json-from-dist/-/package-json-from-dist-1.0.1.tgz#4f1471a010827a86f94cfd9b0727e36d267de505"
