NodeJS中async.retry+fetch结合parallelLimit致程序冻结问题求助

核心冻结原因 #

1. 未消费fetch响应体：node-fetch发起请求后，若不读取响应内容（如res.text()/res.json()），TCP连接不会被释放。当并发请求耗尽系统文件描述符或连接池时，后续请求无法建立，导致所有任务停滞，parallelLimit永远无法完成。
2. 不合理的重试配置：async.retry设置了1000000次重试，一旦某个请求失败，会无限重试，直接导致任务无法结束。
3. 变量泄漏：doStressTest中的result未声明，会成为全局变量，可能引发意外行为。
4. async库版本兼容（潜在）：若使用async v3以下版本，parallelLimit不支持Promise返回值，直接await会导致Promise永远pending。

修复步骤 #

1. 强制消费fetch响应体 #

每次fetch后必须读取响应内容，释放连接：

const res = await fetch(`http://localhost:3000/users/${user}/cards`);
await res.text(); // 根据API返回格式选择res.json()或res.text()

2. 调整重试次数为合理值 #

将async.retry的重试次数从1000000改为3-5次，避免无限重试：

async.retry(5, async () => { /* fetch逻辑 */ })

3. 正确使用async.parallelLimit #

确保使用async v3+版本，此时parallelLimit在不传入回调时会返回Promise，可直接await。若使用旧版本，需用util.promisify包装：

import { promisify } from 'util';
const parallelLimit = promisify(async.parallelLimit);

4. 修复变量泄漏 #

在doStressTest中为result添加let声明：

let result = await attemptFetches(10);

5. 优化连接池配置（可选） #

针对高并发场景，配置node-fetch的Agent以优化连接复用：

import http from 'http';
const agent = new http.Agent({
    maxSockets: 1000, // 根据系统资源调整
    keepAlive: true
});

// 在fetch中使用agent
fetch(url, { agent })

完整修复代码 #

import async from "async";
import fetch from "node-fetch";
import http from 'http';

// 配置HTTP连接池，优化高并发表现
const agent = new http.Agent({
    maxSockets: 1000,
    keepAlive: true
});

const users = [];
for(let i = 1; i <= 5000; ++i) {
    users.push(i);
}

async function attemptFetches(limit) {
    const tasks = users.map((user) => {
        return async () => {
            // 最多重试5次，避免无限循环
            const res = await async.retry(5, async () => {
                const response = await fetch(`http://localhost:3000/users/${user}/cards`, { agent });
                if (!response.ok) {
                    throw new Error(`Request failed with status ${response.status}`);
                }
                // 必须消费响应体，释放TCP连接
                await response.text();
                return response;
            });
            return res;
        };
    });

    const startTime = performance.now();
    await async.parallelLimit(tasks, limit);
    const endTime = performance.now();
    return endTime - startTime;
}

async function doStressTest() {
    let result = await attemptFetches(10);
    console.log(`Running ${users.length} fetch queries in sets of 10 took ${result} ms`);
    
    result = await attemptFetches(100);
    console.log(`Running ${users.length} fetch queries in sets of 100 took ${result} ms`);
    
    result = await attemptFetches(1000);
    console.log(`Running ${users.length} fetch queries in sets of 1000 took ${result} ms`);
    
    result = await attemptFetches(5000);
    console.log(`Running ${users.length} fetch queries in sets of 5000 took ${result} ms`);
};

const keepAliveInterval = setInterval(() => {}, 1000);

doStressTest().then(() => {
    clearInterval(keepAliveInterval);
    agent.destroy(); // 测试结束后销毁Agent，释放所有连接
});

内容的提问来源于stack exchange，提问作者Nuubles