[译]关于Node.jsstreams你须要晓得的统统

Node.js的stream模块是著名的运用难题，更别说理解了。那如今可以通知你，这些都不是题目了。

多年来，开发人员在那里建立了大批的软件包，其唯一目的就是运用stream运用起来更简朴，然则在这篇文章里，我们专注于引见原生的Node.js Steam Api。

&＃8220;Stream 是Node.js中最好的却最随意马虎被误会的部份&＃8221; &＃8212;&＃8211; Dominic Tarr

Streams是数据的鸠合，就跟数组和字符串一样。不同点就在于Streams可以不是马上就悉数可用，而且不会悉数载入内存。这使得他异常合适处置惩罚大批数据，或许处置惩罚每隔一段时间有一个数据片断传入的状况。

然则，Stream并不仅仅适用于处置惩罚大数据(大块的数据。。。)。运用它，一样也有利于组织我们大代码。就像我们运用管道去和兼并壮大的Linux敕令。在Node.js中，我们也可以做一样的事变。

const grep = ... // A stream for the grep output
const wc = ... // A stream for the wc input
grep.pipe(wc)

Node.js的许多内置模块都完成了Stream接口

上面例子内里的Node.js对象列表包含了可读流和可写流，有一些对象既是可读流也是可写流，像TCP sockets, zlib 和 crypto streams。

注重这些对象是有很亲昵的关联的。当一个客户端的HTTP 响应对象是一个可读流，那末在效劳器端这就是一个可写流。由于在HTTP例子中，我们一般是从一个对象(http.IncomingMessage)读取再写入到别的一个对象(http.ServerResponse)中去。

还要注重，当涉及到子历程时，stdio流（stdin，stdout，stderr）具有逆流范例。这就许可我们异常随意马虎的运用管道从主历程衔接子历程的Streams。

理论都是很好的，但现实究竟是怎样模样的呢？让我们看一些例子树模代码Streams在内存运用方面的比较。

我们先建立一个大文件

const fs = require('fs');
const file = fs.createWriteStream('./big.file');
for(let i=0; i<= 1e6; i++) {
file.write('Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n');
}
file.end();

看看我运用什么建立文件的？一个可写流嘛

fs模块可以经由过程Stream接口来读取和写入文件。在上面的例子中，我们在轮回中经由过程可写流向big.file写入了1百万行数据。

运转上面的代码会天生一个也许400M的文件

这是一个简朴的Node web效劳器，特地为big.file供应效劳：

const fs = require('fs');
const server = require('http').createServer();
server.on('request', (req, res) => {
fs.readFile('./big.file', (err, data) => {
if (err) throw err;
res.end(data);
});
});
server.listen(8000);

当server收到请求，它会运用异步要领fs.readFile处置惩罚这个big file。然则这并不代表我们会打断事宜轮回机制。一切都是准确的吗？？

那如今当我们启动server，看看内存监视器都发作了什么。

如今接见这个效劳器，看看内存的运用状况。

内存占用马上飙升到434.8 MB。

在我们把文件内容输出到客户端之前，我们就把全部文件读入了内存。这是很低效的。

HTTP response对象(上文中的res对象)也是一个可写流，这就意味着假如我们有一个代表着big file的可读流，我们可以经由过程管道把他们俩衔接起来完成一样的功用，而不需要运用400M内存。

Node的fs模块给我们供应了一个可以操纵任何文件的可读流,经由过程createReadStream要领建立。我们可以把它和response对象衔接起来。

const fs = require('fs');
const server = require('http').createServer();
server.on('request', (req, res) => {
const src = fs.createReadStream('./big.file');
src.pipe(res);
});
server.listen(8000);

如今再去接见server的时候，使人惊奇的事变发作了(看内存监视器)

发作了什么？

当我们接见效劳器的时候，我们经由过程流每次运用一段数据，这意味着我们不是把悉数的数据都加载到内存中。内存运用量只上升了不到25M。

可以把上面的例子用到极致，天生5百万行数据而不是1百万行。这模样的话，这个文件的大小会凌驾2GB，这实际上大于Node中的默许缓冲区限定。

假如你想在server上运用fs.readFile,这在默许状况下是行不通的，除非你改了Node.js的默许缓冲区限定。然则运用fs.createReadStream，把2 GB的数据返回给客户端基础不存在题目，以至内存运用量都没有任何变化。

准备好进修Steam了吗？

在Node.js中有4中基础的流范例：Readable, Writable, Duplex, and Transform streams。

• Readable 可读流是可以从中斲丧数据的源的笼统，一个例子就是fs.createReadStream要领
• Writable 可写流是可以写入数据的目的的笼统，一个例子就是fs.createWriteStream要领
• duplex Steam是一个同时具有读写功用的流，一个例子就是TCP socket
• Transform 是一个双工流，它可以在交换数据的时候做转换。一个例子就是zlib.createGzip运用gzip紧缩数据。你可以把Transform streams当做是一个传入可读流，返回一个可写流的函数。它另有一个别号through streams

一切的Stream都是EventEmitter的实例对象。当流读和写的时候都邑触发响应的事宜。然则另有一个更简朴的运用要领，那就是运用pipe。

The pipe method

要记着下面这个魔幻要领

readableSrc.pipe(writableDest)

在这一行内里，我们经由过程管道把可读流(源)输出到一个可写流内里去(目的)，源必需是一个可写流，目的必需是可写流。固然，他们也都可以是duplex/Transform。现实上，当我们运用管道衔接流的时候，我们可以像在linux中一样运用链式衔接。

readableSrc
.pipe(transformStream1)
.pipe(transformStream2)
.pipe(finalWrtitableDest)

pipe要领返回目的流，这保证了我们可以运用链式挪用。关于streams a(可读流)，b,c(可读可写流)，d可写流，我们可以运用：

a.pipe(b).pipe(c).pipe(d)
# Which is equivalent to:
a.pipe(b)
b.pipe(c)
c.pipe(d)
# Which, in Linux, is equivalent to:
$ a | b | c | d

pipe要领是运用流最轻便的要领。一般经由过程管道和事宜的要领运用流，然则要只管防止二者混用。一般当你运用pipe要领就不需要运用事宜了。然则当你需要更多定制的操纵的话，运用事宜的体式格局会更好。

Stream events

除了从可读流读取数据传输到可写流，pipe要领还自动处置惩罚一些其他事变。比方处置惩罚毛病，处置惩罚文件完毕操纵，流之间速率快慢题目。

同时，流也可以直接运用事宜操纵。以下是和管道相称的经由过程事宜操纵流的要领。

# readable.pipe(writable)
readable.on('data', (chunk) => {
writable.write(chunk);
});
readable.on('end', () => {
writable.end();
});


下面是一些主要流的事宜和要领。

这些事宜和要领在某种程度上是相干的，由于它们一般被一同运用。

可读流上的最主要的事宜是

• data事宜，当可读撒布输了一段数据的时候会触发
• end事宜，当没有数据被传输时触发

可写流上的最主要的事宜是

• drain事宜，当可写流可以吸收事宜的时候被触发
• finish事宜，当一切数据被吸收时被触发

事宜和要领可以连系起来，以便定制和优化流的运用。读取可读流，我们可以运用pipe/unpipe要领，或许read/unshift/resume要领。运用可写流，我们可以可写流作为pipe/unpipe要领的参数，或许运用write要领写入，运用end要领封闭。

可读流的停息和活动

可读流有两个很主要的形式影响了我们运用的体式格局。

• 停息形式
• 活动形式

这些形式有时候被称为拉和推形式

一切的可读流最先的时候都是默许停息形式，然则它们可以随意马虎的被切换成活动形式，当我们需要的时候又可以切换成停息形式。有时候这个切换是自动的。

当一个可读流是停息形式的时候，我们可以运用read要领从流中读取。然则当一个流是活动形式的时候，数据是延续的活动，我们需要运用事宜去监听数据的变化。

在活动形式中，假如可读流没有监听者，可读流的数据会丧失。这就是为何当可读流逝活动形式的时候，我们必需运用data事宜去监听数据的变化。现实上，只需增加一个数据事宜处置惩罚递次即可将停息的流转换为流形式，删除数据事宜处置惩罚递次将流切换回停息形式。 个中一些是为了与旧的Node Stream接口举行向后兼容。

可以运用resume()和pause()要领在这两种形式之间切换。

当我们运用pipe要领操纵可读流的时候是不需要忧郁上面的这些操纵的，由于pipe要领会自动帮我们处置惩罚这些题目。

当我们议论Node.js中的流时，有两项主要的使命：

• 流的建立
• 流的运用

我们到如今为止议论的都是怎样运用流，那下面来看看怎样建立吧！

Streams的建立一般运用stream模块。

建立一个可写流

为了建立一个可写流，我们需要运用stream模块内里的Writable类。

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

我们可以有许多种体式格局完成一个可写流。比方，我们可以继续Writable类。

class myWritableStream extends Writable {
}

然则我更喜好运用组织函数的体式格局建立。经由过程给Writable通报一些参数来建立一个对象。唯一必需要传的选项时write要领，它需要暴漏需要写入的数据块。

const { Writable } = require('stream');
const outStream = new Writable({
write(chunk, encoding, callback) {
console.log(chunk.toString());
callback();
}
});
process.stdin.pipe(outStream);

write要领吸收3个参数

• chunk一般是一个buffer对象，我们可以经由过程设置修正
• encoding在这类状况下就需要了，不过一般状况是可以疏忽的
• callback是当我们处置惩罚完这个数据块的时候需要挪用的函数。这是一个写入是不是胜利的信号。假如失利了，给这个回调通报一个Error对象

在outStream中，我们简朴的把chunk打印出来，由于并没有发作毛病，我们直接挪用了callback要领。这是这是简朴并不有效的打印流。它会打印吸收到的一切值。

为了运用这个流，我们可以简朴的process.stdin这个可读流。经由过程pipe要领衔接起来。

当我们运转上面的例子，任何我们在控制台输入的内容都邑被console.log打印出来。

这不是一个异常有效的流的完成，然则它已被Node.js内置完成了。outStream功用和process.stdout基础相似。我们也可以经由过程pipe要领把stdin和stdout衔接起来并完成一样的功用。

process.stdin.pipe(process.stdout);

建立一个可读流

建立可读流，我们需要Readable类

const { Readable } = require('stream');
const inStream = new Readable({});

建立一个可读流异常简朴。可以运用push要领推入数据给其他流运用

const { Readable } = require('stream');
const inStream = new Readable();
inStream.push('ABCDEFGHIJKLM');
inStream.push('NOPQRSTUVWXYZ');
inStream.push(null); // No more data
inStream.pipe(process.stdout);

当我们push一个null对象进去的时候，这就标志着我们要停止传输了。

我们可以简朴的把这个流经由过程pipe要领衔接到一个可写流process.stdout

运转上面的代码，会猎取一切的inStream的数据并打印出来。异常简朴但有效。

我们在经由过程pipe衔接之前，就会把一切的数据推送到流内里。更好的要领是在消费者请求时按需推送数据。可以经由过程修正可读流设置内里的read()要领完成。

const inStream = new Readable({
read(size) {
// there is a demand on the data... Someone wants to read it.
}
});

当读取要领在可读流上被挪用时，该完成可以将部份数据推送到行列。 比方，我们可以一次推一个字母，从字符代码65（示意A）最先，并在每次推送时递增：

const inStream = new Readable({
read(size) {
this.push(String.fromCharCode(this.currentCharCode++));
if (this.currentCharCode > 90) {
this.push(null);
}
}
});
inStream.currentCharCode = 65;
inStream.pipe(process.stdout);

当有流在读取可读流的数据的时候，read要领会延续实行，如许就会一向推出更多的字符。我们需要在某个时候停止它，这就是为何我们设置了一个停止前提推入了null。

我们应当一直按需推送数据。

Duplex/Transform 流的完成

运用Duplex流，我们经由过程同一个对象完成可读流和可写流。这相似同时完成了两个接口。

下面这个例子就连系了上面两个可读流和可写流的综合例子。

const { Duplex } = require('stream');
const inoutStream = new Duplex({
write(chunk, encoding, callback) {
console.log(chunk.toString());
callback();
},
read(size) {
this.push(String.fromCharCode(this.currentCharCode++));
if (this.currentCharCode > 90) {
this.push(null);
}
}
});
inoutStream.currentCharCode = 65;
process.stdin.pipe(inoutStream).pipe(process.stdout);

经由过程兼并这些要领，我们可以运用这个duplex流读取从A-Z的字母也一样可以运用它的打印功用。我们把stdin流衔接到这个duplex上去运用它的打印功用，再把这个duplex流自身衔接到stdout上去就在控制台看到了A-Z。

双工流的可读写的两侧完整自力运转。就像一个对象上两种自力的功用。

transform流是一种更风趣的duplex流。由于它的输出来源于她的输入。

关于一个transform流，我们不需要完成read和write要领，我们仅仅需要完成transform要领，这个要领兼并了它们两个。它具有写入要领的功用，也可以用它推送数据。

这是一个简朴的transform例子，把任何输入转换成大写。

const { Transform } = require('stream');
const upperCaseTr = new Transform({
transform(chunk, encoding, callback) {
this.push(chunk.toString().toUpperCase());
callback();
}
});
process.stdin.pipe(upperCaseTr).pipe(process.stdout);

在这个transformstream内里，像上个例子中双工流一样。然则我们只完成了transform()要领。我们把chunk转换成大写，再把大写字母作为可读流的输入。

Streams Object Mode

默许，流会吸收 Buffer/String 范例的数据。另有个字段 objectMode 设置，可以让stream 吸收恣意范例的对象。

下面是一个这类范例的例子。以下变更流的组合使得将逗号分开值的字符串映照为Javascript对象的功用。 所以“a，b，c，d”成为{a：b，c：d}。

const { Transform } = require('stream');
const commaSplitter = new Transform({
readableObjectMode: true,
transform(chunk, encoding, callback) {
this.push(chunk.toString().trim().split(','));
callback();
}
});
const arrayToObject = new Transform({
readableObjectMode: true,
writableObjectMode: true,
transform(chunk, encoding, callback) {
const obj = {};
for(let i=0; i obj[chunk[i]] = chunk[i+1];
}
this.push(obj);
callback();
}
});
const objectToString = new Transform({
writableObjectMode: true,
transform(chunk, encoding, callback) {
this.push(JSON.stringify(chunk) + '\n');
callback();
}
});
process.stdin
.pipe(commaSplitter)
.pipe(arrayToObject)
.pipe(objectToString)
.pipe(process.stdout)

我们经由过程commasplitter通报输入字符串（比方，“a，b，c，d”），它将数组作为其可读数据（[“a”，“b”，“c”，“d”]））。 在该流上增加可读的ObjectMode标志是必要的，由于我们正在将对象推送到其上，而不是字符串。

然后我们把数组导入到arrayToObject数据流中，我们需要把writableObjectMode设置为 true，以示意arrayToObject会吸收一个对象。别的它还会推送一个对象出去，所以还要把他的readableObjectMode为true。末了一个objectToString吸收一个对象然则输出字符串，所以就只需要设置一个writableObjectMode。

Node.js内置transform streams对象

Node有一些异常有效的内置transform streams对象。这包含zlib和crypto。

下面这个例子运用了zlib.createGzip()连系了额fs readable/writable streams完成了文件紧缩。

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];
fs.createReadStream(file)
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream(file + '.gz'));

你可以运用上面的剧本紧缩任何你传入的参数文件。我们把文件的可读撒布入了zlib的内置转换流。再写入到新的.gz文件中。

运用管道另有一个很酷的事变，就是可以和事宜连系起来。比方我想用户看到进度，并在完毕的时候发个音讯。由于pipe要领会返回目的流，我们也可以经由过程链式注册事宜。

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];
fs.createReadStream(file)
.pipe(zlib.createGzip())
.on('data', () => process.stdout.write('.'))
.pipe(fs.createWriteStream(file + '.zz'))
.on('finish', () => console.log('Done'));

所以运用管道要领，我们可以轻松地操纵流，然则我们还可以运用需要的事宜进一步定制与这些流的交互。

管道要领的优点是，我们可以用它来以一种可读的体式格局一一组成我们的递次。 比方，我们可以简朴地建立一个变更流来报告进度，而不必监听上面的数据事宜，并用另一个.pipe()挪用替代 .on() 挪用：

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];
const { Transform } = require('stream');
const reportProgress = new Transform({
transform(chunk, encoding, callback) {
process.stdout.write('.');
callback(null, chunk);
}
});
fs.createReadStream(file)
.pipe(zlib.createGzip())
.pipe(reportProgress)
.pipe(fs.createWriteStream(file + '.zz'))
.on('finish', () => console.log('Done'));

reportProgress流是一个简朴的pass-through流，然则也跟规范事宜一样报告进度。注重callback()函数的第二个参数，这相当于把数据推送出去。

连系流的运用是无止境的。比方，假如我们需要在我们gzip之前或以后加密文件，我们需要做的就是根据我们需要的确实递次来治理另一个转换流。运用Node的crypto模块处置惩罚这个事变。

const crypto = require('crypto');
// ...
fs.createReadStream(file)
.pipe(zlib.createGzip())
.pipe(crypto.createCipher('aes192', 'a_secret'))
.pipe(reportProgress)
.pipe(fs.createWriteStream(file + '.zz'))
.on('finish', () => console.log('Done'));

上面的剧本紧缩然后加密通报的文件，只要具有暗码的人材可以运用文件。 我们没法运用一般的解紧缩有效递次解紧缩此文件，由于它已被加密。

为了可以解紧缩文件，我们需要运用完整相反的操纵，这也很简朴。

fs.createReadStream(file)
.pipe(crypto.createDecipher('aes192', 'a_secret'))
.pipe(zlib.createGunzip())
.pipe(reportProgress)
.pipe(fs.createWriteStream(file.slice(0, -3)))
.on('finish', () => console.log('Done'));

假定通报的文件是紧缩版本，上面的代码将建立一个读取流，将其传输到crypto createDecipher（）流中（运用雷同的隐秘），将其输出管道输入到zlib createGunzip（）流中， 然后将文件写回到没有扩展名的文件中。

以上就是悉数了，感谢浏览！！

翻译自Node.js Streams: Everything you need to know

建立了一个递次员交换微信群，人人进群交换IT手艺

假如已过期，可以增加博主微信号15706211347，拉你进群