Node.js 中的流(Stream)是出了名的难用甚至是难以理解。
用 dominic Tarr 的话来说:“流是 Node 中最好的,也是最容易被误解的想法。”即使是 Redux 的创建者和 react.js 的核心团队成员 Dan Abramov 也害怕 Node 流。
本文将帮助你了解流以及如何使用。不要害怕,你完全可以把它搞清楚!
流是为 Node.js 应用提供动力的基本概念之一。它们是数据处理方法,用于将输入的数据顺序读取或把数据写入输出。
流是一种以有效方式处理读写文件、网络通信或任何类型的端到端信息交换的方式。
流的处理方式非常独特,流不是像传统方式那样将文件一次全部读取到存储器中,而是逐段读取数据块并处理数据的内容,不将其全部保留在内存中。
这种方式使流在处理大量数据时非常强大,例如,文件的大小可能大于可用的内存空间,从而无法将整个文件读入内存进行处理。那是流的用武之地!
既能用流来处理较小的数据块,也可以读取较大的文件。
以 YouTube 或 Netflix 之类的“流媒体”服务为例:这些服务不会让你你立即下载视频和音频文件。取而代之的是,你的浏览器以连续的块流形式接收视频,从而使接收者几乎可以立即开始观看和收听。
但是,流不仅涉及处理媒体和大数据。它们还在代码中赋予了我们“可组合性”的力量。考虑可组合性的设计意味着能够以某种方式组合多个组件以产生相同类型的结果。在 Node.js 中,可以通过流在其他较小的代码段中传递数据,从而组成功能强大的代码段。
与其他数据处理方法相比,流基本上具有两个主要优点:
如果你已经使用过 Node.js,则可能遇到过流。例如在基于 Node.js 的 HTTP 服务器中,request 是可读流,而 response 是可写流。你可能用过 fs 模块,该模块可让你用可读和可写文件流。每当使用 Express 时,你都在使用流与客户端进行交互,而且由于 TCP 套接字、TLS栈和其他连接都基于 Node.js,所以在每个可以使用的数据库连接驱动的程序中使用流。
首先需要可读性流,然后将其初始化。
const Stream = require('stream')
const readableStream = new Stream.Readable()
现在,流已初始化,可以向其发送数据了:
readableStream.push('ping!')
readableStream.push('pong!')
强烈建议在使用流时配合异步迭代器(async iterator)。根据 Axel Rauschmayer 博士的说法,异步迭代是一种用于异步检索数据容器内容的协议(这意味着当前“任务”可以在检索项目之前被暂停)。另外必须提及的是,流异步迭代器实现使用内部的 readable 事件。
从可读流中读取时,可以使用异步迭代器:
import * as fs from 'fs';
async function logChunks(readable) {
for await (const chunk of readable) {
console.log(chunk);
}
}
const readable = fs.createReadStream(
'tmp/test.txt', {encoding: 'utf8'});
logChunks(readable);
// Output:
// 'This is a test!\n'
也可以用字符串收集可读流的内容:
import {Readable} from 'stream';
async function readableToString2(readable) {
let result = '';
for await (const chunk of readable) {
result += chunk;
}
return result;
}
const readable = Readable.from('Good morning!', {encoding: 'utf8'});
assert.equal(await readableToString2(readable), 'Good morning!');
注意,在这种情况下必须使用异步函数,因为我们想返回 Promise。
请切记不要将异步功能与 EventEmitter 混合使用,因为当前在事件处理程序中发出拒绝时,无法捕获拒绝,从而导致难以跟踪错误和内存泄漏。目前的最佳实践是始终将异步函数的内容包装在 try/catch 块中并处理错误,但这很容易出错。 这个 pull request 旨在解决一旦其落在 Node 核心上产生的问题。
要了解有关异步迭代的 Node.js 流的更多信息,请查看这篇很棒的文章。
stream.Readable.from(iterable, [options]) 这是一种实用方法,用于从迭代器中创建可读流,该迭代器保存可迭代对象中包含的数据。可迭代对象可以是同步可迭代对象或异步可迭代对象。参数选项是可选的,除其他作用外,还可以用于指定文本编码。
const { Readable } = require('stream');
async function * generate() {
yield 'hello';
yield 'streams';
}
const readable = Readable.from(generate());
readable.on('data', (chunk) => {
console.log(chunk);
});
根据 Streams API,可读流有效地以两种模式之一运行:flowing和paused。可读流可以处于对象模式,无论处于 flowing 模式还是 paused 模式。
在 flowing 模式中,要从流中读取数据,可以监听数据事件并附加回调。当有大量数据可用时,可读流将发出一个数据事件,并执行你的回调。看下面的代码片段:
var fs = require("fs");
var data = '';
var readerStream = fs.createReadStream('file.txt'); //Create a readable stream
readerStream.setEncoding('UTF8'); // Set the encoding to be utf8.
// Handle stream events --> data, end, and error
readerStream.on('data', function(chunk) {
data += chunk;
});
readerStream.on('end',function() {
console.log(data);
});
readerStream.on('error', function(err) {
console.log(err.stack);
});
console.log("Program Ended");
函数调用 fs.createReadStream() 给你一个可读流。最初流处于静态状态。一旦你侦听数据事件并附加了回调,它就会开始流动。之后将读取大块数据并将其传递给你的回调。流实现者决定发送数据事件的频率。例如,每当有几 KB 的数据被读取时,HTTP 请求就可能发出一个数据事件。当从文件中读取数据时,你可能会决定读取一行后就发出数据事件。
当没有更多数据要读取(结束)时,流将发出结束事件。在以上代码段中,我们监听此事件以在结束时得到通知。
另外,如果有错误,流将发出并通知错误。
在 paused 模式下,你只需在流实例上重复调用 read(),直到读完所有数据块为止,如以下示例所示:
var fs = require('fs');
var readableStream = fs.createReadStream('file.txt');
var data = '';
var chunk;
readableStream.on('readable', function() {
while ((chunk=readableStream.read()) != null) {
data += chunk;
}
});
readableStream.on('end', function() {
console.log(data)
});
read() 函数从内部缓冲区读取一些数据并将其返回。当没有内容可读取时返回 null。所以在 while 循环中,我们检查是否为 null 并终止循环。请注意,当可以从流中读取大量数据时,将会发出可读事件。
所有 Readable 流均以 paused 模式开始,但可以通过以下方式之一切换为 flowing 模式:
Readable 可以使以下方法之一切换回 paused 模式:
一个需要记住的重要概念是,除非提供了一种用于消耗或忽略该数据的机制,否则 Readable 将不会生成数据。如果使用机制被禁用或取消,则 Readable 将会试图停止生成数据。添加 readable 事件处理会自动使流停止 flowing,并通过 read.read() 得到数据。如果删除了 readable 事件处理,那么如果存在 'data' 事件处理,则流将再次开始 flowing。
要将数据写入可写流,你需要在流实例上调用 write()。如以下示例所示:
var fs = require('fs');
var readableStream = fs.createReadStream('file1.txt');
var writableStream = fs.createWriteStream('file2.txt');
readableStream.setEncoding('utf8');
readableStream.on('data', function(chunk) {
writableStream.write(chunk);
});
上面的代码很简单。它只是简单地从输入流中读取数据块,并使用 write() 写入目的地。该函数返回一个布尔值,指示操作是否成功。如果为 true,则写入成功,你可以继续写入更多数据。如果返回 false,则表示出了点问题,你目前无法写任何内容。可写流将通过发出 drain 事件来通知你什么时候可以开始写入更多数据。
调用 writable.end() 方法表示没有更多数据将被写入 Writable。如果提供,则可选的回调函数将作为 finish 事件的侦听器附加。
// Write 'hello, ' and then end with 'world!'.
const fs = require('fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed!
你可以用可写流从可读流中读取数据:
const Stream = require('stream')
const readableStream = new Stream.Readable()
const writableStream = new Stream.Writable()
writableStream._write = (chunk, encoding, next) => {
console.log(chunk.toString())
next()
}
readableStream.pipe(writableStream)
readableStream.push('ping!')
readableStream.push('pong!')
writableStream.end()
还可以用异步迭代器来写入可写流,建议使用
import * as util from 'util';
import * as stream from 'stream';
import * as fs from 'fs';
import {once} from 'events';
const finished = util.promisify(stream.finished); // (A)
async function writeIterableToFile(iterable, filePath) {
const writable = fs.createWriteStream(filePath, {encoding: 'utf8'});
for await (const chunk of iterable) {
if (!writable.write(chunk)) { // (B)
// Handle backpressure
await once(writable, 'drain');
}
}
writable.end(); // (C)
// Wait until done. Throws if there are errors.
await finished(writable);
}
await writeIterableToFile(
['One', ' line of text.\n'], 'tmp/log.txt');
assert.equal(
fs.readFileSync('tmp/log.txt', {encoding: 'utf8'}),
'One line of text.\n');
stream.finished() 的默认版本是基于回调的,但是可以通过 util.promisify() 转换为基于 Promise 的版本(A行)。
在此例中,使用以下两种模式:
Writing to a writable stream while handling backpressure (line B):
在处理 backpressure 时写入可写流(B行):
if (!writable.write(chunk)) {
await once(writable, 'drain');
}
关闭可写流,并等待写入完成(C行):
writable.end();
await finished(writable);
管道是一种机制,可以将一个流的输出作为另一流的输入。它通常用于从一个流中获取数据并将该流的输出传递到另一个流。管道操作没有限制。换句话说,管道可用于分多个步骤处理流数据。
在 Node 10.x 中引入了 stream.pipeline()。这是一种模块方法,用于在流转发错误和正确清理之间进行管道传输,并在管道完成后提供回调。
这是使用管道的例子:
const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');
// 使用 pipeline api 可以轻松将一系列流
// 通过管道传输在一起,并在管道完全完成后得到通知。
// 一个有效地用 gzip压缩巨大视频文件的管道:
pipeline(
fs.createReadStream('The.Matrix.1080p.mkv'),
zlib.createGzip(),
fs.createWriteStream('The.Matrix.1080p.mkv.gz'),
(err) => {
if (err) {
console.error('Pipeline failed', err);
} else {
console.log('Pipeline succeeded');
}
}
);
由于pipe 不安全,应使用 pipeline 代替 pipe。
Node.js 流模块 提供了构建所有流 API 的基础。
Stream 模块是 Node.js 中默认提供的原生模块。 Stream 是 EventEmitter 类的实例,该类在 Node 中异步处理事件。因此流本质上是基于事件的。
要访问流模块:
const stream = require('stream');
stream 模块对于创建新型流实例非常有用。通常不需要使用 stream 模块来消耗流。
由于它们的优点,许多 Node.js 核心模块提供了原生流处理功能,最值得注意的是:
以下是与可写流相关的一些重要事件:
这就是所有关于流的基础知识。流、管道和链是 Node.js 的核心和最强大的功能。流确实可以帮你编写简洁而高效的代码来执行 I/O。
另外,还有一个值得期待的 Node.js 战略计划,称为 BOB,旨在改善 Node.js 的内部数据流以及希望作为未来 Node.js 流数据接口的公共 API 的。
关于 Node.js 里 ES6 Modules 的一次更新说明,总结来说:CommonJS 与 ES6 Modules 之间的关键不同在于代码什么时候知道一个模块的结构和使用它。
在这个教程中,我们会开发一个命令行应用,它可以接收一个 CSV 格式的用户信息文件,教程的内容大纲:“Hello,World”,处理命令行参数,运行时的用户输入,异步网络会话,美化控制台的输出,封装成 shell 命令,JavaScript 之外
首先你需要生成https证书,可以去付费的网站购买或者找一些免费的网站,可能会是key或者crt或者pem结尾的。不同格式之间可以通过OpenSSL转换
nodej项目在微信环境开发,nodejs的异步特效,会导致请求没有完成就执行下面的代码,出现错误。经过多方查找,可以使用async模块来异步转同步,只有前一个function执行callback,下一个才会执行。
3G的大文件分1500个2M二进度文件,通post方法发送给node服务,服务器全部接收到文件后,进组装生成你上文件。
JavaScript比C的开发门槛要低,尽管服务器端JavaScript存在已经很多年了,但是后端部分一直没有市场,JavaScript在浏览器中有广泛的事件驱动方面的应用,考虑到高性能、符合事件驱动、没有历史包袱这3个主要原因,JavaScript成为了Node的实现语言。
node.js的第一个基本论点是I / O的性能消耗是很昂贵。因此,使用当前编程技术的最大浪费来自于等待I / O完成。有几种方法可以处理性能影响
在前后端分离的开发中,通过 Restful API 进行数据交互时,如果没有对 API 进行保护,那么别人就可以很容易地获取并调用这些 API 进行操作。那么服务器端要如何进行鉴权呢?
我们经常跟Node.js打交道,即使你是一名前端开发人员 -- npm脚本,webpack配置,gulp任务,程序打包 或 运行测试等。即使你真的不需要深入理解这些任务,但有时候你会感到困惑,会因为缺少Node.js的一些核心概念而以非常奇怪的方式来编码。
运行在 Node.js 之上的 Webpack 是单线程模型的,也就是说 Webpack 需要处理的任务需要一件件挨着做,不能多个事情一起做。happypack把任务分解给多个子进程去并发的执行,子进程处理完后再把结果发送给主进程。
内容以共享、参考、研究为目的,不存在任何商业目的。其版权属原作者所有,如有侵权或违规,请与小编联系!情况属实本人将予以删除!