[NODE之14]Stream

偏执的太偏执、 2022-06-18 02:12 128阅读 0赞

/**
     * Created by liyanq on 17/3/27.
     * 流（stream）在 Node.js 中是处理流数据的抽象接口（abstract interface）。
     * stream 模块提供了基础的 API 。使用这些 API 可以很容易地来构建实现流接口的对象。
     * Node.js 提供了多种流对象。 例如， HTTP 请求 和 process.stdout 就都是流的实例。
     * 流可以是可读的、可写的，或是可读写的。所有的流都是 EventEmitter 的实例。
     *
     * HTTP 请求是流的实例～
     */
    
    /*流的类型
     * Node.js 中有四种基本的流类型：
     * Readable - 可读的流 (例如 fs.createReadStream()).
     * Writable - 可写的流 (例如 fs.createWriteStream()).
     * Duplex - 可读写的流 (例如 net.Socket).//这个得注意～～
     * Transform - 在读写过程中可以修改和变换数据的 Duplex 流 (例如 zlib.createDeflate()).*/
    
    /*对象模式
     * 所有使用 Node.js API 创建的流对象都只能操作 strings 和 Buffer 对象
     * 在创建流的实例时，可以通过 objectMode 选项使流的实例切换到对象模式。试图将已经存在的流切换到对象模式是不安全的。*/
    
    /*缓冲
     * Writable 和 Readable 流都会将数据存储到内部的缓存（buffer）中。
     * 这些缓存可以 通过相应的 writable._writableState.getBuffer() 或 readable._readableState.buffer来获取。*/
    
    /*流消费者的 API
     * req 是 http.IncomingMessage 的实例，这是一个 Readable Stream
     * res 是 http.ServerResponse 的实例，这是一个 Writable Stream*/
    
    const util = require("util");
    const fs = require("fs");
    const Stream = require("stream");
    
    /*Stream.Readable可读流
     * 可读流（Readable stream）接口是对你正在读取的数据的来源的抽象。
     * 换句话说，数据来来自可读流（Readable stream）不会分发数据，直到你表明准备就绪。
     * 可读流（Readable stream） 有2种模式: 流动模式（flowing mode） 和 暂停模式（paused mode）.
     * 流动模式（flowing mode）时,尽快的从底层系统读取数据并提供给你的程序。
     * 暂停模式（paused mode）时, 你必须明确的调用 stream.read() 来读取数据。
     * 暂停模式（paused mode） 是默认模式。
     * 注意: 如果没有绑定数据处理函数，并且没有 pipe() 目标，流会切换到流动模式（flowing mode），并且数据会丢失。
     *
     * 可以通过下面几个方法，将流切换到流动模式（flowing mode）。
     * 添加一个 'data' 事件 事件处理器来监听数据.  (就这一个就够用了～～)
     * 调用 resume() 方法来明确的开启数据流。
     * 调用 pipe() 方法来发送数据给Writable.
     *
     * 可以通过以下方法来切换到暂停模式（paused mode）:
     * 如果没有 导流（pipe） 目标，调用 pause()方法.
     * 如果有 导流（pipe） 目标, 移除所有的 'data' 事件处理函数, 调用 unpipe() 方法移除所有的 导流（pipe） 目标。
     * 注意, 为了向后兼容考虑， 移除 'data' 事件监听器并不会自动暂停流。
     * 同样的，当有导流目标时，调用  pause() 并不能保证流在那些目标排空后，请求更多数据时保持暂停状态。
     *
     * http responses, on the client
     * http requests, on the server
     * fs read streams
     * zlib streams
     * crypto streams
     * tcp sockets
     * child process stdout and stderr
     * process.stdin
     *
     * 事件
     * 'readable':当一个数据块可以从流中读出，将会触发'readable' 事件.`
     *            注意:一旦内部缓存排空，一旦有更多数据将会再次触发 readable 事件。
     * 'close':将在流或其底层资源（比如一个文件）关闭后触发。'close' 事件触发后，该流将不会再触发任何事件。
     *         不是所有 Readable 都会触发 'close' 事件。
     *  'data':chunk <Buffer> | <String> | <any> 数据片段。对于非对象模式的可读流，这是一个字符串或者 Buffer。
     *         当流转换到 flowing 模式时会触发该事件
     *  'end':将在流中再没有数据可供消费时触发。
     *  'error':事件可以在任何时候在可读流实现（Readable implementation）上触发。
     *          通常，这会在底层系统内部出错从而不能产生数据，或当流的实现试图传递错误数据时发生
     *  'readable':将在流中有数据可供读取时触发。与data事件冲突？？
     *             注意： 通常情况下， 应该使用 readable.pipe() 方法和 'data' 事件机制，而不是 'readable' 事件。
     *
     *方法
     * readable.isPaused():大多数情况下，没有必要直接使用该方法。
     * readable.pause():readable.pause() 方法将会使 flowing 模式的流停止触发 'data' 事件， 进而切出 flowing 模式。
     *
     * readable.pipe(destination[, options]):
     * destination <stream.Writable> 数据写入目标
     * options <Object> Pipe 选项 end <Boolean> 在 reader 结束时结束 writer 。默认为 true。
     *
     * readable.read([size])
     * size <Number> Optional argument to specify how much data to read.
     * Return <String> | <Buffer> | <Null>
     *
     * readable.resume()
     * Returns: this；会触发"data"事件
     *
     * readable.setEncoding(encoding)
     * readable.unpipe([destination])*/
    
    const options = {
        //要是想读取一个文件的特定区域，可以配置start、end属性，指定起始和结束（包含在内）的字节偏移
        flags: "r",      //’w’代表写，’r’代表读，类似的还有’r+’、’w+’、’a’等，与Linux下的open函数接受的读写模式类似。
        encoding: "utf8",//默认就是“utf8”，你还可以为它指定”ascii”或”base64”
        mode: 666,//?
        autoClose: true  //autoClose属性为true（默认行为）时，当发生错误或文件读取结束时会自动关闭文件描述符。
    };
    
    var readableStream = fs.createReadStream("stream1", options);
    var writeAbleStream = fs.createWriteStream("我是被写入的～", {flags: "w", encoding: "utf-8"});
    
    /*继承路径:readableStream:ReadStream->stream.Readable->events.EventEmitter implements NodeJS.ReadableStream*/
    readableStream.setEncoding("utf-8");//指定了编码，收到的将是字符串～
    readableStream.pipe(writeAbleStream, {end: false});//直接流入到了可写流，并且覆盖了可写流内容，还创建了文件～
    
    /*如果在这里删除管道的话，writeAbleStream里面什么也没写，而且readableStream也暂停了～
     readableStream.unpipe(writeAbleStream);
     console.log(readableStream.isPaused());//true
     readableStream.resume();*/
    
    setTimeout(function () {
        readableStream.unpipe(writeAbleStream);
        writeAbleStream.end();
    }, 200);//200ms写了6000多行～～～
    
    
    readableStream.on("data", function (data) {
        console.log("------------------------------------------------");
        console.log(data);
        readableStream.pause();//暂停的话，就不触发"end"、"close"事件
        setTimeout(function () {
            readableStream.resume();//恢复后继续...
        }, 1000)
    });
    
    readableStream.on("close", function () {//不是所有Readable都会触发 'close' 事件。
        console.log("readableStream的close事件")
    });
    
    readableStream.on("error", function (err) {//没试验出来～～
        console.log("readableStream的error事件" + util.inspect(err));
    });
    
    readableStream.on("end", function () {//比close要早触发～。
        console.log("readableStream的end事件")
    });
    
    /*可写流
     * export interface WritableStream extends EventEmitter {
     * writable: boolean;
     * isTTY?: boolean;
     * write(buffer: Buffer | string, cb?: Function): boolean;
     * write(str: string, encoding?: string, cb?: Function): boolean;
     * end(): void;
     * end(buffer: Buffer, cb?: Function): void;
     * end(str: string, cb?: Function): void;
     * end(str: string, encoding?: string, cb?: Function): void;}
     *
     * 'close':将在流或其底层资源（比如一个文件）关闭后触发。'close' 事件触发后，该流将不会再触发任何事件。
     *         不是所有可写流都会触发 'close' 事件。
     * 'drain':如果调用 stream.write(chunk) 方法返回 false，流将在适当的时机触发 'drain' 事件，
     *         这时才可以继续向流中写入数据。(这点很重要～～)
     * 'error':在写入数据出错或者使用管道出错时触发。事件发生时，回掉函数仅会接收到一个 Error 参数。
     *         注意: 'error' 事件发生时，流并不会关闭。
     * 'finish':在调用了 stream.end() 方法，且缓冲区数据都已经传给底层系统（underlying system）之后， 'finish' 事件将被触发。
     *
     * 'pipe':src <stream.Readable> 输出到目标可写流（writable）的源流（source stream）
     *        const writer = getWritableStreamSomehow();
     *        const reader = getReadableStreamSomehow();
     *        writer.on('pipe', (src) => {
     *            console.error('something is piping into the writer');
     *            assert.equal(src, reader);
     *        });
     *        reader.pipe(writer);
     *
     * 'unpipe':src <Readable Stream> unpiped 当前可写流的源流
     *          const writer = getWritableStreamSomehow();
     *          const reader = getReadableStreamSomehow();
     *          writer.on('unpipe', (src) => {
     *               console.error('Something has stopped piping into the writer.');
     *               assert.equal(src, reader);
     *          });
     *          reader.pipe(writer);
     *          reader.unpipe(writer);
     *
     * writable.end([chunk][, encoding][, callback]):
     *          chunk <String> | <Buffer> | <any> 可选的，需要写入的数据。对于非对象模式下的流， chunk 必须是字符串或者 Buffer。
     *          对于对象模式下的流， chunk 可以是任意的 JavaScript 值，除了 null。
     *          encoding <String> 如果 chunk 是字符串，这里指定字符编码。
     *          callback <Function> 可选的，流结束时的回调函数
     *          调用了 stream.end() 方法之后，再调用 stream.write() 方法将会导致错误。
     * writable.setDefaultEncoding(encoding):encoding <String> 新的默认编码 返回： this
     *
     * writable.cork():调用 writable.cork() 方法将强制所有写入数据都内存中的缓冲区里。(没试验~)
     * writable.uncork():writable.uncork() 将输出在 stream.cork() 方法被调用之后缓冲在内存中的所有数据。
     *         stream.cork();
     *         stream.write('some ');
     *         stream.cork();
     *         stream.write('data ');
     *         process.nextTick(() => {
     *               stream.uncork();
     *               // 之前的数据只有在 uncork() 被二次调用后才会输出
     *               stream.uncork();
     *         });
     * writable.write(chunk[, encoding][, callback]):
     *          chunk <String> | <Buffer> 要写入的数据
     *          encoding <String> 如果 chunk 是字符串，这里指定字符编码
     *          callback <Function> 缓冲数据输出时的回调函数
     *          返回： <Boolean> 如果流需要等待 'drain' 事件触发才能继续写入数据，这里将返回 false ； 否则返回 true。
     *          如果有错误发生， callback 不一定 会接收到这个错误作为第一个参数。要确保可靠地检测到写入错误，应该监听 'error' 事件。
     *
     * Writable 的例子包括了：
     * HTTP requests, on the client
     * HTTP responses, on the server
     * fs write streams
     * zlib streams
     * crypto streams
     * TCP sockets
     * child process stdin
     * process.stdout, process.stderr*/
    
    var stream4 = fs.createWriteStream("stream4");
    function write(data, cb) {
        if (!stream4.write(data)) {
            console.log("drain");
            stream4.once('drain', cb)
        } else {
            console.log("nextTick");
            process.nextTick(cb)
        }
    }
    
    write('hello world', () => {
        console.log('write completed, do more writes now')
    });
    
    /*----------------------------------------------------------------------------------*/
    /*流拷贝完整例子
     * 文件一共:1352(MB);已经完成1352(MB);完成百分比:100%;拷贝速度:48(MB/S)
     * 共用时间:2.507(s)
     */
    const filmReadStream = fs.createReadStream("/Users/liyanq/Downloads/buffer.rmvb", {flags: "r"});
    const filmWriteStream = fs.createWriteStream("拷贝的电影.rmvb");
    
    var stat = fs.statSync("/Users/liyanq/Downloads/buffer.rmvb");
    var totalSizeMB = Math.ceil(stat.size / (1024 * 1024));
    var passedSizeByte = 0;
    var lastSizeMB = 0;
    var startTime = Date.now();
    
    filmReadStream.on("data", function (data) {
        passedSizeByte += Buffer.byteLength(data);
        if (!filmWriteStream.write(data)) {
            filmReadStream.pause();
        }
    });
    
    filmReadStream.on("end", function () {
        filmWriteStream.end();
    });
    
    filmWriteStream.on("drain", function () {
        filmReadStream.resume();
    });
    
    var interval = setInterval(function () {
        var passedSizeMB = Math.ceil(passedSizeByte / (1024 * 1024));
        var percent = Math.ceil((passedSizeMB / totalSizeMB) * 100);
        var notPassedSizeMb = totalSizeMB - passedSizeMB;
        var speedSizeMb = passedSizeMB - lastSizeMB;
        lastSizeMB = passedSizeMB;
        console.log("文件一共:%d(MB);已经完成%d(MB);完成百分比:%d%;拷贝速度:%d(MB/S)",
            totalSizeMB, passedSizeMB, percent, speedSizeMb * 2);
        if (notPassedSizeMb <= 0) {
            console.log("共用时间:%d(s)", (Date.now() - startTime) / 1000);
            clearInterval(interval);
        }
    }, 500);
    
    /*Duplex 流与 Transform 流
    * stream.Duplex 类#
    * Duplex 流是同时实现了 Readable 和 Writable 接口的流。
    * Duplex 流的实例包括了：
    * TCP sockets(重要)
    * zlib streams
    * crypto streams
    *
    * stream.Transform 类
    * 变换流（Transform streams） 是一种 Duplex 流。它的输出与输入是通过某种方式关联的。
    * 和所有 Duplex 流一样，变换流同时实现了 Readable 和 Writable 接口。
    * 变换流的实例包括：
    * zlib streams
    * crypto streams*/

new stream.Readable的方式没试验成功；以后遇到再说吧。

http://nodejs.cn/api/stream.html\#stream\_duplex\_and\_transform\_streams

http://www.tuicool.com/articles/meuuqy