使用Node.js express-session为特定用户限制带宽的问题

带宽分配规则 #

• 无session用户：10Mbps（上传+下载双向）
• 已登录且用户为foo：25Mbps
• 已登录且用户为bar：50Mbps

遇到的问题 #

尝试stream-throttle、node-throttle、express-bandwidth-throttle等库实现限速，结果：

1. 限速完全失效，仍能全速传输
2. 使用busboy处理上传时，文件仅传输1-2MB就中断

之前的错误实现代码：

const { Throttle, ThrottleGroup } = require("stream-throttle");

const app = express();
const port = 8000;

app.use((req, res, next) => {
    req.pipe(new Throttle({ rate: 5 * 1024 * 1024 / 8 })); //5Mbps
    next();
});

上传处理代码（busboy）：

function uploadBusboy(req, res, next) {
console.log('User using Busboy Uploader Method...');

req.pipe(req.busboy); // Pipe it through busboy here

let targetFolderC = '';
let folderName = '';
let numberOfFiles = 0;
let totalFile = 0;
let tempFiles = [];

req.busboy.on('field', (fieldname, val) => {
    if (fieldname === 'targetFolder') {
        targetFolderC = val;
    }
    if (fieldname === 'folderName') {
        folderName = val;
    }
});

req.busboy.on('file', (fieldname, file, filename) => {
    console.log(`Upload of '${filename.filename}' started`);
    numberOfFiles++;
    totalFile++;
    var tempPath = path.join("./public/Uploader/uploadsTemp/", filename.filename);

    tempFiles.push(tempPath);
    const fstream = fs.createWriteStream(tempPath);
    file.pipe(fstream);

    fstream.on('close', () => {
        console.log(`Upload of '${filename.filename}' finished`);
        numberOfFiles--;
        if (numberOfFiles <= 0) {
            tempFiles.forEach((temp) => {
                const targetFolder = path.join("./public/Uploader/uploads/", targetFolderC, folderName);
                const oldPath = temp;
                const newPath = path.join(targetFolder, temp.substring(temp.lastIndexOf("uploadsTemp") + 12));
                fs.mkdirSync(targetFolder, { recursive: true });
                fs.renameSync(oldPath, newPath);
            });
            console.log(`Received total of ${totalFile} files`);
            res.redirect('back');
        }    
    });
});

核心问题分析 #

你之前的写法完全无效，根源：

1. 仅把req pipe到节流器，但未将节流后的流替换回req——后续busboy仍读取原始未节流的请求流，等于白做。
2. 未区分上传（请求数据流）和下载（响应数据流），只处理了req，下载自然不会限速。

步骤1：实现动态带宽计算 #

先写一个函数根据用户session返回对应带宽（注意：stream-throttle的rate单位是字节/秒，1Mbps = 128KB/s = 131072字节/秒）：

// 根据session返回节流速率（字节/秒）
function getThrottleRate(req) {
    let mbps = 10; // 默认无session用户
    if (req.session?.user) {
        switch(req.session.user) {
            case 'foo':
                mbps = 25;
                break;
            case 'bar':
                mbps = 50;
                break;
        }
    }
    return mbps * 1024 * 1024 / 8; // Mbps转字节/秒
}

步骤2：上传节流（处理请求流） #

不能直接req.pipe(throttle)，需要把节流流替换为req的可读流，确保后续中间件（包括busboy）使用节流后的流：

const { Throttle } = require("stream-throttle");

// 该中间件必须放在express-session之后，busboy之前
app.use((req, res, next) => {
    // 仅对有请求体的方法（POST/PUT/PATCH）做上传节流
    if (['POST', 'PUT', 'PATCH'].includes(req.method)) {
        const throttle = new Throttle({ rate: getThrottleRate(req) });
        // 将原始请求流导向节流器
        req.pipe(throttle);
        // 替换req的可读流接口，让后续逻辑读取节流后的流
        req.readable = true;
        req.read = (...args) => throttle.read(...args);
        req.on = (...args) => throttle.on(...args);
        req.off = (...args) => throttle.off(...args);
        req.emit = (...args) => throttle.emit(...args);
    }
    next();
});

步骤3：下载节流（处理响应流） #

拦截res.write和res.end，让响应数据先经过节流器：

app.use((req, res, next) => {
    const originalWrite = res.write;
    const originalEnd = res.end;
    const throttle = new Throttle({ rate: getThrottleRate(req) });

    // 把节流器的输出导向原始响应
    throttle.pipe(res);

    // 替换res的写入方法，让数据先经过节流
    res.write = (chunk, encoding, callback) => {
        return throttle.write(chunk, encoding, callback);
    };

    res.end = (chunk, encoding, callback) => {
        return throttle.end(chunk, encoding, callback);
    };

    next();
});

步骤4：适配Busboy上传 #

你的busboy代码无需大改，因为我们已经替换了req的可读流接口，直接保持原有的req.pipe(req.busboy)即可正常工作：

// 确保该路由在节流中间件之后加载
app.post('/upload', uploadBusboy);

额外注意事项 #

1. 必须确保express-session中间件在节流中间件之前加载，否则req.session无法正常获取。
2. 本地测试可能看不出效果（磁盘IO瓶颈会掩盖带宽限制），请用真实网络环境测试。
3. 如果需要统一管理同一用户的多个连接，可以改用ThrottleGroup。

内容的提问来源于stack exchange，提问作者Khỉ Em