[node.js] [HTTP/S] 实现 requests 发起 HTTP/S/1.1/2.0 请求

node.js 使用 V8 引擎来编译运行 javascript 代码，与浏览器中的环境不同的是，node.js 不包含 DOM 和 BOM 模块。

本文使用 node.js 的官方库来实现一个简单的 requests() 函数，可以用来发送 HTTP/1.1 和 HTTP/2.0 的请求。有关 HTTP/1.1 和 HTTP/2.0 请参见往期的文章 HTTP 版本的演进。

在 node.js http2 默认支持 keep-alive 连接，使用 http2 来发起 HTTP 请求需要我们自己来管理 client (TCP 连接)。

思路：创建一个 TCPConnection 类，用来保存 client 对象。创建一个 ConnectionPool 类，用来自动管理连接池，并定期清理长期没有请求的 client 对象。 requests() 函数可以控制 HTTP 的版本 HTTP/1.1 还是 HTTP/2.0，也可以选择 GET 或者 POST 方法，接收的数据经过解压缩后和响应头一起封装到对象中进行返回。

代码如下：（代码中加入了足够多的注释）

// requests.js
import http from 'node:http';
import https from 'node:https';
import http2 from 'node:http2';
import zlib from 'node:zlib';



// http2.connect 创建的 client 实例维护者一个 TCP 连接
// 将 client 封装成类，实现 TCP 连接的复用
class TCPConnection {  
    constructor(client, origin) {  
        this.client = client;  
        this.origin = origin;  // 记录 client 的 origin: https://example.com:port
        this.expires = Date.now() + 2 * 60 * 1000; // 设置过期时间为2分钟后  
  
        // 监听 error 事件，打印错误信息，关闭 client 
        this.client.on('error', (err) => {  
            console.error(`Client error for origin ${this.origin}:`, err);  
            this.close(); // 在出现错误时关闭连接  
        });  
  
        // 监听 client 的 close() 事件，打印 origin
        this.client.on('close', () => {  
            console.log(`Connection closed for origin ${this.origin}`);  
            // 连接关闭后，不需要再次关闭，因为 this.client.close() 已经被调用或者即将被调用  
            // 但可以更新任何相关的状态或日志  
        });  
    }

    isExpired() {  
        return Date.now() > this.expires;     // 返回 client 是否过期
    }  

    close() {  
        this.client.close();     // 用来关闭过期的 client，过多的 client 会消耗系统的资源
    }
} 


// 维护一个连接池，保存所有带有 client 的 TCPConnection 类的实例
class ConnectionPool {  
    constructor() {  
        this.connections = new Map(); // 使用 Map 来存储连接，以便按 origin 查找  
        this.checkInterval = setInterval(() => this.checkExpiredConnections(), 3 * 60 * 1000); // 每3分钟检查一次  
    }  
  
    // 添加一个带有 origin 的连接  
    addConnection(origin) {  
        const client = http2.connect(origin);  
        const connection = new TCPConnection(client, origin);  
        this.connections.set(origin, connection); // 使用 origin 作为键  
        return connection;  
    }  
  
    // 根据 origin 获取连接  
    getConnection(origin) {  
        let connection = this.connections.get(origin);  
  
        if (connection && !connection.isExpired()) {  
            console.log(`使用现有的连接: ${connection.origin}`);      // 测试是否使用了现有的连接
            connection.expires = Date.now() + 2 * 60 * 1000;     // 连接被重新使用，重置 expires 过期时间
            return connection.client; // 返回现有的 client  
        } else {  
            // 如果连接不存在或已过期，则创建新连接  
            connection = this.addConnection(origin);  
            return connection.client;  
        }  
    }  
  
    checkExpiredConnections() {  
        for (const [origin, connection] of this.connections) {  
            if (connection.isExpired()) {  
                connection.close(); // 关闭过期的连接  
                this.connections.delete(origin); // 从 Map 中移除  
            }  
        }  
    }  
  
    closeAll() {     // 程序结束后调用，关闭所有的连接
        for (const connection of this.connections.values()) {  
            connection.close();  
        }
        this.connections.clear(); // 清空 Map  
        clearInterval(this.checkInterval); // 清除定期检查  
    }  
}


const conPool = new ConnectionPool();



/**
 * 解压缩数据
 * @param {Buffer} data - 要解压缩的数据
 * @param {string} encoding - 数据的编码方式
 * @returns {Promise<Buffer>} - 解压缩后的数据
 */
async function decompressData(data, encoding) {
    return new Promise((resolve, reject) => {
        switch (encoding) {
            case 'gzip':
                zlib.gunzip(data, (err, decoded) => {
                    if (err) reject(err);
                    else resolve(decoded);
                });
                break;
            case 'deflate':
                zlib.inflate(data, (err, decoded) => {
                    if (err) reject(err);
                    else resolve(decoded);
                });
                break;
            case 'br':
                zlib.brotliDecompress(data, (err, decoded) => {
                    if (err) reject(err);
                    else resolve(decoded);
                });
                break;
            /*
            case 'zstd':
                break;     //以后实现
            */
            default:
                resolve(data); // 如果内容未经任何编码或者压缩，亦或者是图片、视频，直接返回原始数据
        }
    });
}


/**
 * 调用前确定好 http 的版本，向 http/1.1 的服务器发送 2.0 的请求会报 Protocol Error 的错误。
 * @param {string} url - 想要请求的 url
 * @param {string} method - 想要使用的方法 'GET' / 'POST' 等
 * @param {string} httpVersion -  控制 http 的版本: ['1.1' | '2.0']
 * @param {object} headers - request headers 请求头
 * @param {Buffer|string} [data] - 适用于 POST 方法（可选）
 * @returns {Promise<{data: Buffer, headers: object}>} - 返回响应数据和 headers 的 Promise
 */
export default async function requests(url, method, httpVersion, headers, data) {
    return new Promise((resolve, reject) => {0
        try {
            const reqUrl = new URL(url);
            /*
                const myURL = new URL('https://example.com:8080/path?query=param#hash');
                // 访问各个部分
                console.log(myURL.origin);   // "https://example.com:8080"
                console.log(myURL.protocol); // "https:"
                console.log(myURL.hostname);  // "example.com"
                console.log(myURL.port);      // "8080"
                console.log(myURL.pathname);  // "/path"
                console.log(myURL.search);    // "?query=param"
                console.log(myURL.hash);      // "#hash"
            */
            //console.log(reqUrl);
            if ( httpVersion === '1.1' ) {     // 使用 http/1.1 发出请求，node:http、node:https 版本都是 1.1
                const options = {
                    hostname: reqUrl.hostname,
                    port: reqUrl.port || (reqUrl.protocol === 'http:' ? 80 : 443),
                    method: method,
                    path: `${reqUrl.pathname}${reqUrl.search}`,
                    headers: headers
                };
                //console.log(options);
    
                const request = (reqUrl.protocol === 'http:' ? http : https).request(options, (response) => {
                    var resData = Buffer.alloc(0);     //创建一个大小为 0 字节的 Buffer 实例
    
                    response.on('data', (chunk) => {
                        resData = Buffer.concat([resData, chunk]);   // 将resData和新的数据块chunk合并

                    });
    
                    response.on('end', () => {
                        decompressData(resData, response.headers['content-encoding'])    // 检索响应头的 content-encoding 字段进行响应的解码操作
                            .then((decodedData) => {
                                resolve({ data: decodedData, headers: response.headers });
                                // 如果数据经过了 gzip / deflate / br 压缩，就执行解压操作再返回
                                // 数据 和 响应头 封装到对象中一起返回，返回响应头的必要性：方便后续的查看响应头，以进行一些操作
                            })
                            .catch((err) => {
                                reject(`Decompression error: ${err.message}`);
                            });
                    });
                });
    
                request.on('error', (e) => {
                    reject(`Problem with request: ${e.message}`);
                });
    
                if (method === 'POST' && data) {
                    // Ensure the data is a Buffer or convert it
                    request.write(Buffer.isBuffer(data) ? data : Buffer.from(data));
                }
    
                request.end();
            } else if ( httpVersion === '2.0' ) {
                const client = conPool.getConnection(reqUrl.origin);
                /*
                const client = http2.connect(reqUrl.origin);
                http2.connect 会为每个连接创建新的 TCP 连接。如果想要重用连接，需要使用相同的 client 实例。
                http2 不允许 connection: keep-alive 。 因为 http2 本身就设计为支持持久连接的。
                这意味着，HTTP/2 连接自动保持打开状态，以便在同一连接上处理多个请求和响应。
                */
                client.on('error', (err)=> { 
                    reject(`Create client error: ${err}`);
                });
                const options = {
                    ':authority': reqUrl.host,
                    ':method': method,
                    ':path': `${reqUrl.pathname}${reqUrl.search}`,
                    ':scheme': reqUrl.protocol.slice(0, -1), // 去掉末尾的 ':'
                    ...headers   // headers 所有属性合并到 options 中
                };
                /*
                // 将 headers 中的属性合并到 options 中，除了 ...headers，也可以使用以下语句：
                Object.assign(options, headers);
                */
                //console.log(options);
                const request = client.request(options);
                var resHeaders;
                request.on('response', (headers) => {
                    resHeaders = headers;
                });
                var resData = Buffer.alloc(0);
                request.on('data', (chunk) => {
                    resData = Buffer.concat([resData, chunk]);
                });
                request.on('end', () => {
                    decompressData(resData, resHeaders['content-encoding'])
                        .then((decodeData) => {
                            resolve({ data: decodeData, headers: resHeaders});
                        })
                        .catch((err) => {
                            reject(`Decompression error: ${err.message}`);
                        });
                    // client.close(); 使用 ConnectionPool 类来管理 client
                    /*
                    当所有请求完成后，调用 client.close() 以关闭连接。只有在不再需要连接时才应该关闭。
                    client.close(); 会关闭 TCP 连接，新的 http 请求将无法重用 TCP 连接。
                    为了有效地重用连接，应保持 client 对象的引用，以便后续请求可以使用同一连接。
                    可以在一个更大的作用域中定义 client，并在多个请求中复用。
                    */
                });
                request.on('error', (e) => {
                    reject(`Problem with request: ${e.message}`);
                    // client.close();
                });
                if (method === 'POST' && data) {
                    request.write(Buffer.isBuffer(data) ? data : Buffer.from(data));
                }
                request.end();
            } else {
                reject(`Unsupported http version ${httpVersion}`);
            }
        } catch (error) {
            reject(`Invalid URL: ${error.message}`);
        };
    });
}


process.on('exit', () => {
    console.log("执行清理工作: 清理所有的 client 连接 ...");
    conPool.closeAll();
});

在 app.js 中用来发起请求测试，请求图片，和请求我之前发布一段命令行下旋转 cube 的视频：

import requests from './requests/requests.js';
//app.js
//测试
import fs from 'fs';

//请求图片并写入文件
requests('https://www.baidu.com/img/PCtm_d9c8750bed0b3c7d089fa7d55720d6cf.png', 'GET', "1.1", {})
    .then(resObj => {
        console.log(resObj.data);
        console.log(resObj.headers);
        fs.writeFile('baidu.png', resObj.data, {encoding: 'binary'}, error => {   //以二进制写入文件
            if (error) {
                console.error('写入文件时出错:', err);
            } else {
                console.log('文件已成功写入。');
            }
        });
    })
    .catch(error => {
        console.error("Error: ", error);
    });


//请求一段视频
var videos = [
    '32b018315e66b2f02a2c08433b42fcc0_0.ts',
    '32b018315e66b2f02a2c08433b42fcc0_1.ts',
    '32b018315e66b2f02a2c08433b42fcc0_2.ts',
    '32b018315e66b2f02a2c08433b42fcc0_3.ts',
    '32b018315e66b2f02a2c08433b42fcc0_4.ts'
];
var baseUrl = "https://v-blog.csdnimg.cn/asset/999efa6d97215aa8905a1a05f7398e9f/play_video/";

async function downloadAndWriteVideos() {
    for (let index = 0; index < videos.length; index++) {
        let url = baseUrl + videos[index];
        console.log(`index ${index}/${videos.length - 1} : requesting Url: ${url}.`);
        try {
            const resObj = await requests(url, 'GET', "2.0", {});
            //写入每个视频段到单独的文件
            await fs.promises.writeFile('cube_' + index + '.ts', resObj.data, { encoding: 'binary' });
            console.log('数据写入成功: cube_' + index + '.ts');
            
            // 追加到 cube.ts 文件
            await fs.promises.appendFile('cube.ts', resObj.data, { encoding: 'binary' });
            console.log('数据追加成功。');
        } catch (error) {
            console.error("Error: ", error);
        }
    }
}

//调用函数
downloadAndWriteVideos();

process.on('SIGINT', () => {
    console.log("接收到 SIGINT 信号，程序即将退出 ...");
    process.exit();
});

process.on('SIGTERM', () => {
    console.log("接收到 SIGTERM 信号，程序即将退出 ...");
    process.exit();
});