BullMQ搭配node-fetch内存占用过高问题排查求助

1. 失败任务无限制堆积（核心元凶） #

你的队列配置中removeOnFail: false，加上每个任务最多重试5次，所有失败任务（含重试失败的）会永久存放在Redis的失败集合中。如果目标服务不稳定，短时间内就会生成大量失败任务，直接占用Redis内存。

解决办法：
修改队列的defaultJobOptions，给失败任务设置自动过期删除：

new Queue('a-queue', {
    connection: {
        url: process.env.REDIS_URL,
        enableReadyCheck: true,
    },
    defaultJobOptions: {
        attempts: 5,
        removeOnComplete: true,
        removeOnFail: {
            age: 3600 // 失败任务保留1小时后自动删除
        },
    },
})

如果不需要保留失败任务，也可以直接设置removeOnFail: true。

2. 全局并发限制未正确生效 #

你调用的queue.setGlobalConcurrency(5)是针对单个队列实例的，若A、B、C三个队列是分别创建的，需要给每个队列都配置全局并发，否则单个队列的任务并发可能超出预期，导致队列积压、Redis内存占用飙升。

解决办法：
封装队列创建逻辑，确保每个队列都设置全局并发：

// 统一创建队列的函数
const createQueue = async (queueName) => {
    const queue = new Queue(queueName, {
        connection: { url: process.env.REDIS_URL, enableReadyCheck: true },
        defaultJobOptions: {
            attempts: 5,
            removeOnComplete: true,
            removeOnFail: { age: 3600 }
        }
    });
    await queue.setGlobalConcurrency(5);
    return queue;
};

// 创建三个队列
const queueA = await createQueue('a-queue');
const queueB = await createQueue('b-queue');
const queueC = await createQueue('c-queue');

3. Worker线程资源泄漏 #

启用useWorkerThreads: true后，若线程内的资源未正确复用，会导致线程阻塞、任务处理变慢，间接造成队列积压，加重Redis负担。

解决办法：

• 复用fetch的agent实例，避免每次请求新建连接：

// 在a_processor.js顶部初始化agent（复用）
const agentDcIp = new HttpsAgent({ keepAlive: true });

// 后续请求直接复用该agent
const response = await fetch(`${domain}`, {
    headers: headers,
    agent: agentDcIp,
    signal: AbortSignal.timeout(15000) as any,
});

• 复用HeaderGenerator实例，不要每次任务都新建：

// 在a_processor.js顶部初始化一次
const headerGenerator = new HeaderGenerator();

// 任务处理时直接调用生成headers
const headers = headerGenerator.getHeaders();

4. Redis连接复用优化 #

每个Queue和Worker都单独创建Redis连接，会增加Redis的连接数负担，间接导致内存占用上升。

解决办法：
全局复用Redis连接配置：

// 全局定义Redis连接
const redisConn = {
    url: process.env.REDIS_URL,
    enableReadyCheck: true,
};

// 创建队列和Worker时复用该连接
new Queue('a-queue', { connection: redisConn, /* 其他配置 */ });
new Worker('a-queue', processor, { connection: redisConn, /* 其他配置 */ });

临时应急方案 #

如果Redis已经崩溃，可手动清理失败任务释放内存：

# 进入Redis命令行
redis-cli
# 删除BullMQ的失败任务集合
DEL bull:failed

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