Node.jsRedis高效并发与排队实现方法

偷偷努力，悄无声息地变强，然后惊艳所有人！哈哈，小伙伴们又来学习啦~今天我将给大家介绍《Node.js Redis高效并发与排队系统构建指南》，这篇文章主要会讲到等等知识点，不知道大家对其都有多少了解，下面我们就一起来看一吧！当然，非常希望大家能多多评论，给出合理的建议，我们一起学习，一起进步！

本文探讨了如何在Node.js应用中高效管理具有全局并发限制和资源独占时长的特定资源（如“标题”）访问。针对传统队列方案在多资源类型并发场景下的局限性，提出并详细阐述了基于Redis的“火速请求，轮询获取”策略。该方案通过分离请求处理与资源分配，利用Redis的原子操作和过期机制实现资源锁与用户排队，有效解决了长连接阻塞和并发等待问题，提升了系统可伸缩性和用户体验。

引言：并发资源访问的挑战

在Node.js应用中，我们经常面临需要管理有限资源访问的场景。例如，一个系统允许用户请求特定“标题”（如标题A或标题B），但为了维护系统稳定性或业务逻辑，可能存在以下限制：

1. 全局并发限制： 整个应用在某一时刻只能处理一个核心任务。
2. 资源独占与时长： 一旦用户获取到某个标题，将独占该标题一段时间（例如60秒），期间其他用户无法获取，除非当前用户手动释放。
3. 类别隔离： 不同标题的请求应尽量互不影响。例如，用户1请求标题A，用户2请求标题A，用户3请求标题B。理想情况下，用户3（请求标题B）不应长时间等待用户1（请求标题A）的60秒独占期，而只应等待其初始处理阶段。用户2则必须等待用户1释放标题A。

传统的基于内存队列（如p-queue）的解决方案，通过简单地嵌套队列来尝试实现这种复杂逻辑时，往往会遇到非预期的阻塞问题。例如，在上述场景中，用户3可能会错误地被阻塞，直到用户1的标题A独占期结束，这显然违背了类别隔离的初衷，并导致用户体验下降。

这种问题的核心在于，将长时间的资源独占（60秒）与初始的资源获取逻辑耦合在同一个处理流程中，并且试图通过单进程内存队列来管理分布式或长时间的资源状态。为了解决这一挑战，我们需要一种更健壮、可伸缩的方案，将资源状态管理从核心业务处理中分离出来。

核心策略：Redis与“火速请求，轮询获取”

为了克服上述挑战，我们引入了基于Redis的“火速请求，轮询获取”（Fire-and-Forget with Polling）策略。

为什么不直接持有HTTP连接？ 直接在HTTP请求处理过程中持有连接60秒，直到资源释放，存在以下显著问题：

• HTTP超时： 大多数HTTP服务器和客户端都有默认的请求超时时间，长时间持有连接会导致请求失败。
• 资源长期占用： 每个活跃连接都会占用服务器资源，在高并发场景下容易导致资源耗尽。
• 难以扩展： 这种模式限制了并发处理能力，难以水平扩展。

“火速请求，轮询获取”模式 这种模式将用户请求分为两个阶段：

1. 火速请求（Fire-and-Forget）： 用户发起请求后，服务器迅速将请求加入一个等待队列，并立即向客户端返回一个确认信息（例如“您的请求已入队”），而不必等待资源实际分配。
2. 轮询获取（Polling）： 客户端在收到入队确认后，定期（例如每隔几秒）向服务器发送查询请求，询问其请求的资源是否已被分配。服务器根据资源当前状态和队列情况进行判断并响应。

Redis的关键作用 Redis作为高性能的内存数据库，在此模式中扮演了至关重要的角色：

• 分布式锁： 利用Redis的原子操作和过期机制，实现对特定标题的独占锁，确保同一时间只有一个用户持有某个标题。
• 分布式队列： 使用Redis的列表（List）数据结构，为每个标题维护一个独立的等待队列，确保请求按序处理。
• 状态管理： 集中存储和管理所有标题的当前持有者和等待队列信息，使系统具备分布式和高可用的能力。

系统架构与数据模型

本方案的核心在于利用Redis存储和管理标题的独占状态和用户等待队列。

Redis数据结构设计

1. 资源持有者键 (String):

   • 键名格式： title::current_holder
   • 存储内容： 当前持有该标题的用户ID（例如"user123"）。
   • 特性： 设置EX（过期时间，如60秒），实现标题的自动释放。当键过期时，表示标题被释放，可供下一个用户获取。
   • 作用： 实现标题的独占锁。

2. 标题等待队列 (List):

   • 键名格式： title::queue
   • 存储内容： 按照请求顺序存储等待获取该标题的用户ID列表（例如["user2", "user5", "user8"]）。
   • 特性： 使用Redis的列表操作（RPUSH入队，LPOP或LREM出队）维护队列。
   • 作用： 确保同一标题的请求按先进先出（FIFO）的原则进行处理。

整体流程图（文字描述）

1. 用户请求： 客户端向Node.js服务器发送请求，要求获取特定标题。
2. 服务器入队： Node.js服务器接收请求后，立即将用户ID添加到对应标题的Redis等待队列中，并向客户端返回“请求已入队”的响应。
3. 客户端轮询： 客户端收到入队响应后，启动一个定时器，定期向Node.js服务器发送“检查访问权限”的轮询请求。
4. 服务器分配： Node.js服务器在处理轮询请求时，检查Redis中该标题的current_holder状态和queue状态。如果标题未被持有且当前轮询用户是队列中的第一个，则尝试原子性地将该用户设置为标题持有者（设置EX）。
5. 客户端获取： 如果轮询响应表明用户已获得标题，客户端停止轮询，并开始使用该标题。
6. 手动释放： 用户可以随时向服务器发送请求，要求提前释放标题。服务器接收后，删除Redis中的current_holder键。
7. 自动释放： 如果用户在60秒内未手动释放，current_holder键将自动过期，标题被释放。

详细实现步骤

以下是基于Node.js和ioredis库的实现示例。

1. 用户请求标题 (Server-side Endpoint: /request-title)

当用户请求某个标题时，服务器将用户ID添加到对应标题的Redis等待队列中，并立即返回一个确认信息给客户端。

// 假设使用ioredis客户端
import Redis from 'ioredis';
import express from 'express';

const redis = new Redis();
const app = express();
app.use(express.json()); // 用于解析请求体

/**
 * 处理用户请求标题的端点
 * @param {string} req.body.userId - 用户唯一ID
 * @param {string} req.body.titleId - 请求的标题ID (例如 'A', 'B')
 */
app.post('/request-title', async (req, res) => {
    const { userId, titleId } = req.body;
    if (!userId || !titleId) {
        return res.status(400).json({ message: 'User ID and Title ID are required.' });
    }

    const queueKey = `title:${titleId}:queue`;
    // 将用户ID添加到队列末尾
    await redis.rpush(queueKey, userId);

    console.log(`User ${userId} requested title ${titleId}, added to queue.`);
    res.status(202).json({
        message: `Your request for title ${titleId} has been queued.`,
        status: 'queued',
        // 可以返回一个唯一的请求ID供客户端跟踪
    });
});

2. 客户端轮询检查访问权限 (Server-side Endpoint: /check-access)

客户端在收到请求入队响应后，开始定期（例如每5秒）向此端点发送请求。服务器逻辑会检查并尝试分配标题。

/**
 * 处理客户端轮询检查访问权限的端点
 * @param {string} req.query.userId - 用户唯一ID
 * @param {string} req.query.titleId - 请求的标题ID
 */
app.get('/check-access', async (req, res) => {
    const { userId, titleId } = req.query;
    if (!userId || !titleId) {
        return res.status(400).json({ message: 'User ID and Title ID are required.' });
    }

    const holderKey = `title:${titleId}:current_holder`;
    const queueKey = `title:${titleId}:queue`;
    const TITLE_HOLD_SECONDS = 60; // 标题独占时长

    // 1. 检查当前用户是否已持有标题
    const currentHolder = await redis.get(holderKey);
    if (currentHolder === userId) {
        console.log(`User ${userId} already holds title ${titleId}.`);
        return res.json({ status: 'granted', titleId });
    }

    // 2. 如果标题未被持有或已过期，尝试获取
    if (!currentHolder) {
        // 检查队列中第一个用户是否是当前用户
        const [firstInQueue] = await redis.lrange(queueKey, 0, 0);

        if (firstInQueue === userId) {
            // 尝试原子性地设置持有者，并设置过期时间
            // 'NX': Only set the key if it does not already exist.
            // 'EX': Set the specified expire time, in seconds.
            const acquired = await redis.set(holderKey, userId, 'EX', TITLE_HOLD_SECONDS, 'NX');
            if (acquired) {
                // 成功获取，从队列中移除该用户
                await redis.lrem(queueKey, 1, userId);
                console.log(`User ${userId} successfully acquired title ${titleId}.`);
                return res.json({ status: 'granted', titleId });
            }
            // 如果acquired为null，说明在尝试SET NX EX时，有其他进程/线程抢先设置了，
            // 此时当前用户需要继续等待下一次轮询
        }
    }

    // 3. 标题被其他用户持有，或当前用户不在队列首位，继续等待
    console.log(`User ${userId} is waiting for title ${titleId}. Current holder: ${currentHolder || 'None'}`);
    res.json({ status: 'waiting', titleId, currentHolder: currentHolder || null });
});

3. 手动释放标题 (Server-side Endpoint: /abort-title)

用户选择提前释放标题时，客户端发送请求。服务器删除current_holder键。

/**
 * 处理用户手动释放标题的端点
 * @param {string} req.body.userId - 用户唯一ID
 * @param {string} req.body.titleId - 释放的标题ID
 */
app.post('/abort-title', async (req, res) => {
    const { userId, titleId } = req.body;
    if (!userId || !titleId) {
        return res.status(400).json({ message: 'User ID and Title ID are required.' });
    }

    const holderKey = `title:${titleId}:current_holder`;

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~