Node.js Stream.pipeline() Method

Why Stream.pipeline

通过流我们可以将一大块数据拆分为一小部分一点一点的流动起来，而无需一次性全部读入，在 Linux 下我们可以通过 | 符号实现，类似的在 Nodejs 的 Stream 模块中同样也为我们提供了 pipe() 方法来实现。

未使用 Stream pipe 情况

在 Nodejs 中 I/O 操作都是异步的，先用 util 模块的 promisify 方法将 fs.readFile 的 callback 形式转为 Promise 形式，这块代码看似没问题，但是它的体验不是很好，因为它是将数据一次性读入内存再进行的返回，当数据文件很大的时候也是对内存的一种消耗，因此不推荐它。

const Koa = require('koa');
const fs = require('fs');
const app = new Koa();
const { promisify } = require('util');
const { resolve } = require('path');
const readFile = promisify(fs.readFile);app.use(async ctx => {try {ctx.body = await readFile(resolve(__dirname, 'test.json'));} catch(err) { ctx.body = err };
});app.listen(3000);

使用 Stream pipe 情况

下面，再看看怎么通过 Stream 的方式在 Koa 框架中响应数据

...
app.use(async ctx => {try {const readable = fs.createReadStream(resolve(__dirname, 'test.json'));ctx.body = readable;} catch(err) { ctx.body = err };
});

以上在 Koa 中直接创建一个可读流赋值给 ctx.body 就可以了，你可能疑惑了为什么没有 pipe 方法，因为框架给你封装好了，不要被表象所迷惑了，看下相关源码：

// https://github.com/koajs/koa/blob/master/lib/application.js#L256
function respond(ctx) {...let body = ctx.body;if (body instanceof Stream) return body.pipe(res);...
}

没有神奇之处，框架在返回的时候做了层判断，因为 res 是一个可写流对象，如果 body 也是一个 Stream 对象（此时的 Body 是一个可读流），则使用 body.pipe(res) 以流的方式进行响应。

使用 Stream VS 不使用 Stream

动图封面

动图封面

What Stream.pipeline function

The stream.pipeline() method is a module method that is used to the pipe by linking the streams passing on errors and accurately cleaning up and providing a callback function when the pipeline is done. 

Stream.pipeline() 方法是一种模块方法，用于通过链接传递错误的流并在管道完成时准确地清理并提供回调函数来用于管道。

Syntax:

stream.pipeline(...streams, callback)

Parameters: 

This method accepts two parameters as mentioned above and described below.该方法接受如上所述和如下所述的两个参数。

  • …streams: These are two or more streams that are to be piped together.这些是要通过管道连接在一起的两个或多个流。
  • callback: This function is called when the pipeline is fully done and it shows an ‘error’ if the pipeline is not accomplished.当管道完全完成时调用此函数，如果管道未完成，则会显示“错误”。

Return Value:

 It returns a cleanup function. 返回值：它返回一个清理函数。

The below examples illustrate the use of the stream.pipeline() method in Node.js: 

以下示例说明了 Node.js 中的

stream.pipeline() 方法的用法

Example 1: 

// Node.js program to demonstrate the   

// stream.pipeline() method

// Including fs and zlib module

const fs = require('fs');

const zlib = require('zlib');

// Constructing finished from stream

const { pipeline } = require('stream');

// Constructing promisify from

// util

const { promisify } = require('util');

// Defining pipelineAsync method

const pipelineAsync = promisify(pipeline);

// Constructing readable stream

const readable = fs.createReadStream("input.text");

// Constructing writable stream

const writable = fs.createWriteStream("output.text");

// Creating transform stream

const transform = zlib.createGzip();

// Async function

(async function run() {

    try {

        // pipelining three streams

        await pipelineAsync(

            readable,

            transform,

            writable

        );

        console.log("pipeline accomplished.");

    }

    // Shows error

    catch (err) {

        console.error('pipeline failed with error:', err);

    }

})();

Output:

Promise {  }
pipeline accomplished.

Example 2: 

// Node.js program to demonstrate the   

// stream.pipeline() method

// Including fs and zlib module

const fs = require('fs');

const zlib = require('zlib');

// Constructing finished from stream

const { pipeline } = require('stream');

// Constructing promisify from

// util

const { promisify } = require('util');

// Defining pipelineAsync method

const pipelineAsync = promisify(pipeline);

// Constructing readable stream

const readable = fs.createReadStream("input.text");

// Constructing writable stream

const writable = fs.createWriteStream("output.text");

// Creating transform stream

const transform = zlib.createGzip();

// Async function

(async function run() {

    try {

        // pipelining three streams

        await pipelineAsync(

            readable,

            writable,

            transform

        );

        console.log("pipeline accomplished.");

    }

    // Shows error

    catch (err) {

        console.error('pipeline failed with error:', err);

    }

})();

Output: Here, the order of streams is not proper while piping so an error occurs.

Promise {  }
pipeline failed with error: Error [ERR_STREAM_CANNOT_PIPE]: Cannot pipe, not readableat WriteStream.Writable.pipe (_stream_writable.js:243:24)at pipe (internal/streams/pipeline.js:57:15)at Array.reduce ()at pipeline (internal/streams/pipeline.js:88:18)at Promise (internal/util.js:274:30)at new Promise ()at pipeline (internal/util.js:273:12)at run (/home/runner/ThirstyTimelyKey/index.js:33:11)at /home/runner/ThirstyTimelyKey/index.js:45:5at Script.runInContext (vm.js:133:20)

解析Stream.PipeLine

在应用层我们调用了 fs.createReadStream() 这个方法，顺藤摸瓜找到这个方法创建的可读流对象的 pipe 方法实现，以下仅列举核心代码实现，基于 Nodejs v12.x 源码。

2.1.1 /lib/fs.js

导出一个 createReadStream 方法，在这个方法里面创建了一个 ReadStream 可读流对象，且 ReadStream 来自 internal/fs/streams 文件，继续向下找。

// https://github.com/nodejs/node/blob/v12.x/lib/fs.js
// 懒加载，主要在用到的时候用来实例化 ReadStream、WriteStream ... 等对象
function lazyLoadStreams() {if (!ReadStream) {({ ReadStream, WriteStream } = require('internal/fs/streams'));[ FileReadStream, FileWriteStream ] = [ ReadStream, WriteStream ];}
}function createReadStream(path, options) {lazyLoadStreams();return new ReadStream(path, options); // 创建一个可读流
}module.exports = fs = {createReadStream, // 导出 createReadStream 方法...
}

2.1.2 /lib/internal/fs/streams.js

这个方法里定义了构造函数 ReadStream，且在原型上定义了 open、_read、_destroy 等方法，并没有我们要找的 pipe 方法。

但是呢通过 ObjectSetPrototypeOf 方法实现了继承，ReadStream 继承了 Readable 在原型中定义的函数，接下来继续查找 Readable 的实现。

// https://github.com/nodejs/node/blob/v12.x/lib/internal/fs/streams.js
const { Readable, Writable } = require('stream');function ReadStream(path, options) {if (!(this instanceof ReadStream))return new ReadStream(path, options);...Readable.call(this, options);...
}
ObjectSetPrototypeOf(ReadStream.prototype, Readable.prototype);
ObjectSetPrototypeOf(ReadStream, Readable);ReadStream.prototype.open = function() { ... };ReadStream.prototype._read = function(n) { ... };;ReadStream.prototype._destroy = function(err, cb) { ... };
...module.exports = {ReadStream,WriteStream
};

2.1.3 /lib/stream.js

在 stream.js 的实现中，有条注释：在 Readable/Writable/Duplex/... 之前导入 Stream，原因是为了避免 cross-reference(require)，为什么会这样？

第一步 stream.js 这里将 require('internal/streams/legacy') 导出复制给了 Stream。

在之后的 _stream_readable、Writable、Duplex ... 模块也会反过来引用 stream.js 文件，具体实现下面会看到。

Stream 导入了 internal/streams/legacy

上面 /lib/internal/fs/streams.js 文件从 stream 模块获取了一个 Readable 对象，就是下面的 Stream.Readable 的定义。

// https://github.com/nodejs/node/blob/v12.x/lib/stream.js
// Note: export Stream before Readable/Writable/Duplex/...
// to avoid a cross-reference(require) issues
const Stream = module.exports = require('internal/streams/legacy');Stream.Readable = require('_stream_readable');
Stream.Writable = require('_stream_writable');
Stream.Duplex = require('_stream_duplex');
Stream.Transform = require('_stream_transform');
Stream.PassThrough = require('_stream_passthrough');
...

2.1.4 /lib/internal/streams/legacy.js

上面的 Stream 等于 internal/streams/legacy，首先继承了 Events 模块，之后呢在原型上定义了 pipe 方法，刚开始看到这里的时候以为实现是在这里了，但后来看 _stream_readable 的实现之后，发现 _stream_readable 继承了 Stream 之后自己又重新实现了 pipe 方法，那么疑问来了这个模块的 pipe 方法是干嘛的？什么时候会被用？翻译文件名 “legacy=遗留”？有点没太理解，难道是遗留了？有清楚的大佬可以指点下，也欢迎在公众号 “Nodejs技术栈” 后台加我微信一块讨论下！

// https://github.com/nodejs/node/blob/v12.x/lib/internal/streams/legacy.js
const {ObjectSetPrototypeOf,
} = primordials;
const EE = require('events');
function Stream(opts) {EE.call(this, opts);
}
ObjectSetPrototypeOf(Stream.prototype, EE.prototype);
ObjectSetPrototypeOf(Stream, EE);Stream.prototype.pipe = function(dest, options) {...
};module.exports = Stream;

2.1.5 /lib/_stream_readable.js

在 _stream_readable.js 的实现里面定义了 Readable 构造函数，且继承于 Stream，这个 Stream 正是我们上面提到的 /lib/stream.js 文件，而在 /lib/stream.js 文件里加载了 internal/streams/legacy 文件且重写了里面定义的 pipe 方法。

经过上面一系列的分析，终于找到可读流的 pipe 在哪里，同时也更进一步的认识到了在创建一个可读流时的执行调用过程，下面将重点来看这个方法的实现。

module.exports = Readable;
Readable.ReadableState = ReadableState;const EE = require('events');
const Stream = require('stream');ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
ObjectSetPrototypeOf(Readable, Stream);function Readable(options) {if (!(this instanceof Readable))return new Readable(options);...Stream.call(this, options); // 继承自 Stream 构造函数的定义
}
...

2.2 _stream_readable 实现分析

2.2.1 声明构造函数 Readable

声明构造函数 Readable 继承 Stream 的构造函数和原型。

Stream 是 /lib/stream.js 文件，上面分析了，这个文件继承了 events 事件，此时也就拥有了 events 在原型中定义的属性，例如 on、emit 等方法。

const Stream = require('stream');
ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
ObjectSetPrototypeOf(Readable, Stream);function Readable(options) {if (!(this instanceof Readable))return new Readable(options);...Stream.call(this, options);
}

2.2.2 声明 pipe 方法，订阅 data 事件

在 Stream 的原型上声明 pipe 方法，订阅 data 事件，src 为可读流对象，dest 为可写流对象。

我们在使用 pipe 方法的时候也是监听的 data 事件，一边读取数据一边写入数据。

看下 ondata() 方法里的几个核心实现：

  • dest.write(chunk)：接收 chunk 写入数据，如果内部的缓冲小于创建流时配置的 highWaterMark，则返回 true，否则返回 false 时应该停止向流写入数据，直到 'drain' 事件被触发。
  • src.pause()：可读流会停止 data 事件，意味着此时暂停数据写入了。

之所以调用 src.pause() 是为了防止读入数据过快来不及写入，什么时候知道来不及写入呢，要看 dest.write(chunk) 什么时候返回 false，是根据创建流时传的 highWaterMark 属性，默认为 16384 (16kb)，对象模式的流默认为 16。

Readable.prototype.pipe = function(dest, options) {const src = this;src.on('data', ondata);function ondata(chunk) {const ret = dest.write(chunk);if (ret === false) {...src.pause();}}...
};

2.2.3 订阅 drain 事件，继续流动数据

上面提到在 data 事件里，如果调用 dest.write(chunk) 返回 false，就会调用 src.pause() 停止数据流动，什么时候再次开启呢？

如果说可以继续写入事件到流时会触发 drain 事件，也是在 dest.write(chunk) 等于 false 时，如果 ondrain 不存在则注册 drain 事件。

Readable.prototype.pipe = function(dest, options) {const src = this;src.on('data', ondata);function ondata(chunk) {const ret = dest.write(chunk);if (ret === false) {...if (!ondrain) {// When the dest drains, it reduces the awaitDrain counter// on the source.  This would be more elegant with a .once()// handler in flow(), but adding and removing repeatedly is// too slow.ondrain = pipeOnDrain(src);dest.on('drain', ondrain);}src.pause();}}...
};// 当可写入流 dest 耗尽时，它将会在可读流对象 source 上减少 awaitDrain 计数器
// 为了确保所有需要缓冲的写入都完成，即 state.awaitDrain === 0 和 src 可读流上的 data 事件存在，切换流到流动模式
function pipeOnDrain(src) {return function pipeOnDrainFunctionResult() {const state = src._readableState;debug('pipeOnDrain', state.awaitDrain);if (state.awaitDrain)state.awaitDrain--;if (state.awaitDrain === 0 && EE.listenerCount(src, 'data')) {state.flowing = true;flow(src);}};
}// stream.read() 从内部缓冲拉取并返回数据。如果没有可读的数据，则返回 null。在可读流上 src 还有一个 readable 属性，如果可以安全地调用 readable.read()，则为 true
function flow(stream) {const state = stream._readableState;debug('flow', state.flowing);while (state.flowing && stream.read() !== null);
}

2.2.4 触发 data 事件

调用 readable 的 resume() 方法，触发可读流的 'data' 事件，进入流动模式。

Readable.prototype.pipe = function(dest, options) {const src = this;// Start the flow if it hasn't been started already.if (!state.flowing) {debug('pipe resume');src.resume();}...

然后实例上的 resume（Readable 原型上定义的）会在调用 resume() 方法，在该方法内部又调用了 resume_()，最终执行了 stream.read(0) 读取了一次空数据（size 设置的为 0），将会触发实例上的 _read() 方法，之后会在触发 data 事件。

function resume(stream, state) {...process.nextTick(resume_, stream, state);
}function resume_(stream, state) {debug('resume', state.reading);if (!state.reading) {stream.read(0);}...
}

2.2.5 订阅 end 事件

end 事件：当可读流中没有数据可供消费时触发，调用 onend 函数，执行 dest.end() 方法，表明已没有数据要被写入可写流，进行关闭（关闭可写流的 fd），之后再调用 stream.write() 会导致错误。

Readable.prototype.pipe = function(dest, options) {...const doEnd = (!pipeOpts || pipeOpts.end !== false) &&dest !== process.stdout &&dest !== process.stderr;const endFn = doEnd ? onend : unpipe;if (state.endEmitted)process.nextTick(endFn);elsesrc.once('end', endFn);dest.on('unpipe', onunpipe);...function onend() {debug('onend');dest.end();}
}

2.2.6 触发 pipe 事件

在 pipe 方法里面最后还会触发一个 pipe 事件，传入可读流对象

Readable.prototype.pipe = function(dest, options) {...const source = this;dest.emit('pipe', src);...
};

在应用层使用的时候可以在可写流上订阅 pipe 事件，做一些判断，具体可参考官网给的这个示例 stream_event_pipe[1]

2.2.7 支持链式调用

最后返回 dest，支持类似 unix 的用法：A.pipe(B).pipe(C)

Readable.prototype.pipe = function(dest, options) {return dest;
};

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/419132.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

基于WSL的Ubuntu命令行美化

基于WSL的Ubuntu命令行美化

大多数 Linux 发行版中的默认 Shell 是 Bash。Bash 缺乏代码高亮，不易阅读。本文旨在通过安装 Zsh、oh-my-zsh，并应用 Powerlevel10k 主题来解决这一问题。 环境：Windows10 Pro 21H2，OS build: 19044.1766；基于 WSL 的…
阅读更多...
JVM之java内存区域[1]（程序计数器、栈）

JVM之java内存区域[1]（程序计数器、栈）

文章目录 版权声明零 运行时数据区一 程序计数器1.1 加载阶段1.2 执行阶段1.3 多线程情况 二 栈2.1 java虚拟机栈2.2 java虚拟机栈帧的组成2.2.1 局部变量表2.2.2 操作数栈2.2.3 帧数据 2.3 栈内存溢出2.4 设置帧大小2.5 本地方法栈 版权声明 本博客的内容基于我个人学习黑马程…
阅读更多...
Java学习笔记(八)——Lambda表达式

Java学习笔记(八)——Lambda表达式

文章目录 Lambda表达式Lambda表达式的省略写法Lambda练习练习1练习2 算法题算法题1 斐波那契数列算法题2 猴子吃桃子算法题3 爬楼梯 Lambda表达式 Lambda表达式是JDK8开始的一种新语法形式。 基本作用：简化函数式接口的匿名内部类的写法。 注意： Lam…
阅读更多...
Linux中的新建用户、切换用户

Linux中的新建用户、切换用户

目录 一、Linux系统中有哪些用户 二、新建普通用户 三、root账号与普通账号的切换 一、Linux系统中有哪些用户 1.root 超级管理员（不受权限约束） 2.其他用户 普通用户（受到权限约束） 二、新建普通用户 创建新用户 sudo user…
阅读更多...
【数据结构】 顺序表的基本操作 （C语言版）

【数据结构】 顺序表的基本操作 （C语言版）

一、顺序表 1、顺序表的定义： 线性表的顺序存储结构，即将表中的结点按逻辑顺序依次存放在一组地址连续的存储单元里。这种存储方式使得在逻辑结构上相邻的数据元素在物理存储上也是相邻的，可以通过数据元素的物理存储位置来反映其逻辑关系。…
阅读更多...
天龙八部场景编辑器（源码+软件+教程）

天龙八部场景编辑器（源码+软件+教程）

天龙八部场景编辑器，里面包括《源码》，《软件》，《教程》，喜欢研究天龙八部的可以下载看看。 天龙八部场景编辑器（源码软件教程） 下载地址： 链接：https://pan.baidu.com/s/1GWXErav0…
阅读更多...
JAVA 算法介绍（一）

JAVA 算法介绍（一）

1 二分查找 又叫折半查找，要求待查找的序列有序。每次取中间位置的值与待查关键字比较，如果中间位置 的值比待查关键字大，则在前半部分循环这个查找的过程，如果中间位置的值比待查关键字小， 则在后半部分循环这个查…
阅读更多...
UKP3d的管道编辑

UKP3d的管道编辑

山西这家用户在使用UKP3d时，提出以下问题： 1、stp导入的模型怎么测量距离；另外需要把某一个点移动至原点坐标，这个怎么操作呢？ 回复：dist（主要是捕捉点，推荐使用（开启精…
阅读更多...
Nginx实现html页面注入浏览器监控js代码片段

Nginx实现html页面注入浏览器监控js代码片段

一、背景 最近看到关于浏览器监控相关的东西，顺带着就记录一下其实现的大致原理过程。 在我们没对web应用做浏览器监控的时候，我们其实无法感知到用户对我们应用页面的使用习惯、使用中是否遇到问题，例如白屏情况出现多少次、请求失败情况、j…
阅读更多...
【UEFI基础】EDK网络框架（UDP4）

【UEFI基础】EDK网络框架（UDP4）

UDP4 UDP4协议说明 UDP的全称是User Datagram Protocol，它不提供复杂的控制机制，仅利用IP提供面向无连接的通信服务。它将上层应用程序发来的数据在收到的那一刻，立即按照原样发送到网络。 UDP报文格式： 各个参数说明如下&…
阅读更多...
两条链表相同位数相加[中等]

两条链表相同位数相加[中等]

优质博文IT-BLOG-CN 一、题目 给你两个非空的链表，表示两个非负的整数。它们每位数字都是按照逆序的方式存储的，并且每个节点只能存储一位数字。请你将两个数相加，并以相同形式返回一个表示和的链表。你可以假设除了数字0之外，这…
阅读更多...
《WebKit 技术内幕》之五（3）： HTML解释器和DOM 模型

《WebKit 技术内幕》之五（3）： HTML解释器和DOM 模型

3 DOM的事件机制 基于 WebKit 的浏览器事件处理过程：首先检测事件发生处的元素有无监听者，如果网页的相关节点注册了事件的监听者则浏览器会将事件派发给 WebKit 内核来处理。另外浏览器可能也需要处理这样的事件（浏览器对于有些事件必须响应…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023