Node.js多端点限流Parallel queues请求队列偶发阻塞排查

问题背景 #

• 自研API对接代码库运行时偶发请求完全停滞，必须重启服务才能恢复
• 对接的API包含大量配置独立速率限制的动态端点，端点无法提前枚举，且可能在程序运行时发生变更
• 设计目标为实现并行队列系统，保证不同端点的请求完全隔离：单端点触发限流、返回错误、请求量级远高于其他端点时，都不会阻塞其他端点的正常请求。队列作为请求流程的首个入口，不能出现阻塞。

请求处理流程如下：

Queue------|
       |        |
Check Ratelimit-|
       |
  Make Request

流程规则：请求先加入队列；到达队列头部时校验对应端点的速率限制；未超出限制则直接发起请求，超出限制则将该请求重新放回队列等待下次处理。

接口约定：

• 入队调用add(key, value)方法：key为对应端点的哈希值，value存储请求相关数据，包含请求对应Promise的resolve()和reject()回调
• 请求处理完成调用completed(key)方法：传入对应端点的哈希值作为参数
• 判定触发速率限制时调用retryLater(key, pause)方法：pause为距离可以安全发起请求的等待时长（单位：秒）
• 性能要求：最低每秒可处理不少于5个请求

故障现象 #

队列运行时偶发阻塞：部分队列可正常处理请求，部分队列完全停滞。初步判断为突发请求流量触发的竞态条件导致，但无法定位根因。
原有问题实现代码如下：

const EventsEmitter = require("events");

class QueueHandler extends EventsEmitter {
    constructor() {
        super();
        this.queues = {};
    }

    add(key = "null", value) {
        if (!this.queues[key])
            this.queues[key] = { q: new Array(10).fill(null), c: 0, d: 0 };
        this.queues[key].q[this.queues[key].c] = value;
        this.queues[key].c++;
        if (this.queues[key].c > 9)
            this.queues[key].c = 0;
        if (this.queues[key].q.filter(k => k != null).length == 1)
            this.emit("next", key, value);
    }

    completed(key = "null") {
        this.queues[key].q[this.queues[key].d] = null;
        if (this.queues[key].q.filter(k => k != null).length == 0)
            delete this.queues[key];
        else {
            this.queues[key].d++;
            if (this.queues[key].d > 9)
                this.queues[key].d = 0;
            this.emit("next", key, this.queues[key].q[this.queues[key].d]);
        }
    }

    retryLater(key = "null", next = true, pause = 0) {
        this.queues[key].q[this.queues[key].c] = this.queues[key].q[this.queues[key].d];
        this.queues[key].c++;
        if (this.queues[key].c > 9)
            this.queues[key].c = 0;
        this.queues[key].q[this.queues[key].d] = null;
        if (next == true)
            this.next(key);
        else if (pause != 0)
            setTimeout((() => this.next(key)), pause * 1000);
    }
}

module.exports = QueueHandler;

根因定位 #

原有实现从数据结构选择到逻辑设计都存在明显缺陷，阻塞是多个问题叠加触发的：

1. 核心方法缺失
   retryLater逻辑中直接调用this.next(key)，但整个类根本没有定义next实例方法，第一次触发带pause时长的重试逻辑时，就会抛出方法不存在的错误，直接中断事件处理流程，造成队列卡死。
2. 固定长度环形数组无溢出保护
   固定给每个端点分配长度为10的数组做队列，没有做队列满的判断，只要单个端点的待处理请求数超过10个，写入指针c就会直接覆盖还未处理的请求，造成数据丢失，指针逻辑完全混乱。
3. 环形队列指针逻辑错误
   原有代码用c做写入指针、d做读取（队首）指针，但完全没有遵循环形队列的指针校验规则：

• 没有判断空队列/满队列的指针边界，写入指针追上读取指针、读取指针追上写入指针时都没有做对应处理
• retryLater方法直接把队首元素复制到写入指针位置、移动写入指针后将原队首位置设为null，一旦写入指针和读取指针重合，会直接覆盖未处理请求，还会出现队首指针指向null的情况，此时触发next事件传递null值，后续处理逻辑拿到null不会继续执行，队列直接卡死。

4. 竞态条件触发点明确

• 突发流量下，同一个端点短时间内多次调用add，c指针快速回绕，不等next事件处理完成就会覆盖之前排队的请求
• 每次判断队列长度都用filter(k => k!=null).length做全数组遍历，遍历过程中如果有写入/删除操作，拿到的长度值是错误的，会漏触发next事件。

5. 队列销毁逻辑存在漏洞
   completed方法中判断队列长度为0就直接删除对应key的队列对象，此时如果刚好有新的add请求进来，或者retryLater设置的setTimeout刚好触发执行，会直接抛出读取undefined属性的错误，中断事件循环中的处理逻辑，造成队列卡死。

实现思路修正 #

硬套固定长度环形数组完全不符合当前场景需求，动态端点场景下不需要提前预分配固定长度队列，正确实现思路如下：

• 每个端点独立维护一个动态长度的普通数组队列，端点之间完全隔离
• 每个队列维护processing状态锁，保证同一端点同一时间只有一个请求在执行限流校验、发起请求的逻辑，从根源上避免竞态
• 限流需要重试时，将请求放回队尾，等待设置的pause时长结束后再释放锁，尝试处理下一个请求
• 用单独的变量维护队列长度计数，不需要每次遍历数组计算，O(1)复杂度完全满足性能要求，甚至可以支撑远高于每秒5次的请求量
• 队列销毁前做状态校验，避免空指针错误

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