多队列场景下BLPOP致Redis连接阻塞且无法弹出数据

代码中的核心问题 #

• 函数名不匹配：processQueue里调用的popFromQueue并未定义，实际你写的弹出函数是queuePop，这会直接导致无法调用弹出逻辑。
• 异步语法错误：ProcessingFunction是普通函数却使用了await，会抛出语法错误；同时函数内重复声明同名变量data，逻辑完全混乱。
• 无效的返回值等待：processQueue是无限循环的异步函数，永远不会返回结果，所以await processQueue(...)永远拿不到值，后续打印逻辑根本执行不了。
• 队列覆盖问题：代码只处理了单个queueName队列，没有对8个队列分别启动监听进程。

修正后的代码 #

// 从指定队列阻塞弹出数据
async function queuePop(queue) {
    const value = await redis.blpop(queue, 0);
    if (value) {
        return JSON.parse(value[1]);
    }
}

// 单个队列的循环处理逻辑
async function processQueue(queue) {
    while (true) {
        try {
            const data = await queuePop(queue); // 修正函数名调用
            console.log(`Processing data from ${queue}:`, data);
            // 在这里添加你的业务处理逻辑，比如调用实际的业务函数
            // await handleBusinessTask(data);
        } catch (err) {
            console.error(`Error processing ${queue} job`, err);
            // 出错后短暂延迟，避免日志刷屏
            await new Promise(resolve => setTimeout(resolve, 1000));
        }
    }
}

// 启动所有8个队列的处理进程
function startAllQueueProcesses() {
    // 替换成你的8个队列名称
    const queues = ['queue1', 'queue2', 'queue3', 'queue4', 'queue5', 'queue6', 'queue7', 'queue8'];
    queues.forEach(queue => {
        processQueue(queue).catch(err => console.error(`Failed to start ${queue} processor`, err));
    });
}

// 直接启动所有队列处理（不需要setTimeout，除非有初始化前置逻辑）
startAllQueueProcesses();

关键优化说明 #

• 每个队列单独启动一个processQueue实例，8个队列可以同时监听、处理数据，互不阻塞。
• BLPOP的阻塞特性正好适配你“队列几乎始终有数据”的场景，相比LPOP轮询更节省资源。
• 错误处理中添加延迟，避免队列持续出错时疯狂输出日志。
• 简化了冗余的函数嵌套，直接在队列处理循环中完成数据处理，逻辑更清晰。

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