流中的背压
在数据处置惩罚历程当中会涌现一个叫做背压的罕见题目，它形貌了数据传输历程当中缓冲区背面数据的积累，当传输的吸收端具有庞杂的支配时，或许因为某种缘由速度较慢时，来自传入源的数据就有积累的趋向，就像壅塞一样。
要处理这个题目，必需有一个托付体系来确保数据从一个源到另一个源的腻滑活动，差别的社区已针对他们的顺序独专程处理了这个题目，Unix管道和TCP套接字就是很好的例子，而且平常被称为流量掌握，在Node.js中，流是已采纳的处理方案。
本指南的目的是进一步细致申明背压是什么，以及准确流如安在Node.js的源代码中处理这个题目，本指南的第二部份将引见发起的最好实践，以确保在完成流时运用顺序的代码是平安的和优化的。
我们假定你对Node.js中背压、Buffer和EventEmitter的平常定义以及Stream的一些履历有所相识。假如你还没有浏览这些文档，那末起首检察API文档并非一个坏主意，因为它有助于在浏览本指南时扩大你的明白。
数据处置惩罚的题目
在盘算机体系中，数据经由过程管道、sockets和信号从一个历程传输到另一个历程，在Node.js中，我们找到了一种名为Stream的相似机制。流很好！他们为Node.js做了许多事变，险些内部代码库的每一个部份都运用该模块，作为开发人员，我们勉励你运用它们！
const readline = require('readline');
// process.stdin and process.stdout are both instances of Streams
const rl = readline.createInterface({
input: process.stdin,
output: process.stdout
});
rl.question('Why should you use streams? ', (answer) => {
console.log(`Maybe it's ${answer}, maybe it's because they are awesome! :)`);
rl.close();
});经由过程比较Node.js的Stream完成的内部体系东西，可以证实为何经由过程流完成背压机制是一个很好的优化的一个很好的例子。
在一种状况下，我们将运用一个大文件（约〜9gb）并运用熟习的zip(1)东西对其举行紧缩。
$ zip The.Matrix.1080p.mkv虽然这须要几分钟才完成，但在另一个shell中我们可以运转一个剧本，该剧本采纳Node.js的模块zlib，它包括另一个紧缩东西gzip(1)。
const gzip = require('zlib').createGzip();
const fs = require('fs');
const inp = fs.createReadStream('The.Matrix.1080p.mkv');
const out = fs.createWriteStream('The.Matrix.1080p.mkv.gz');
inp.pipe(gzip).pipe(out);要测试效果，请尝试翻开每一个紧缩文件，zip(1)东西紧缩的文件将关照你文件已损坏，而Stream完成的紧缩将无毛病地解紧缩。
注重：在此示例中，我们运用.pipe()将数据源从一端获取到另一端，然则，请注重没有附加准确的毛病处置惩罚顺序。假如没法准确吸收数据块，Readable源或gzip流将不会被烧毁，pump是一个实用东西，假如个中一个流失利或封闭，它将准确地烧毁管道中的一切流，而且在这类状况下是必需的！
只需Nodejs 8.x或更早版本才须要pump，关于Node 10.x或更高版本，引入pipeline来替代pump。这是一个模块要领，用于在流传输之间转发毛病和准确清算，并在管道完成时供应回调。
以下是运用管道的示例：
const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');
// Use the pipeline API to easily pipe a series of streams
// together and get notified when the pipeline is fully done.
// A pipeline to gzip a potentially huge video file efficiently:
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');
}
}
);你还可以在管道上挪用promisify以将其与async/await一同运用：
const stream = require('stream');
const fs = require('fs');
const zlib = require('zlib');
const pipeline = util.promisify(stream.pipeline);
async function run() {
try {
await pipeline(
fs.createReadStream('The.Matrix.1080p.mkv'),
zlib.createGzip(),
fs.createWriteStream('The.Matrix.1080p.mkv.gz'),
);
console.log('Pipeline succeeded');
} catch (err) {
console.error('Pipeline failed', err);
}
}太多的数据，太快
有些状况下，Readable流可以会过快地为Writable供应数据 — 远远凌驾斲丧者可以处置惩罚的数据！
当发作这类状况时，斲丧者将最先列队一切数据块以供今后斲丧，写入行列将变得越来越长，因而在全部历程完成之前，必需将更多数据保留在内存中。
写入磁盘比从磁盘读取要慢许多，因而，当我们尝试紧缩文件并将其写入我们的硬盘时，将发作背压，因为写入磁盘将没法跟上读取的速度。
// Secretly the stream is saying: "whoa, whoa! hang on, this is way too much!"
// Data will begin to build up on the read-side of the data buffer as
// `write` tries to keep up with the incoming data flow.
inp.pipe(gzip).pipe(outputFile);这就是背压机制很主要的缘由，假如没有背压体系，该历程会耗尽体系的内存，有效地减缓了其他历程，并独有你体系的大部份直到完成。
这致使了一些事变：
- 减缓一切其他当前历程。
- 一个异常超负荷的渣滓网络器。
- 内存耗尽。
鄙人面的示例中，我们将掏出.write()函数的返回值并将其变动成true，这有效地禁用了Node.js核心中的背压支撑，在任何对’modified’二进制文件的援用中，我们正在议论在没有return ret;行的状况下运转node二进制，而改成return true;。
渣滓网络器上的过分负荷
我们来看看疾速基准测试，运用上面的雷同示例，我们举行频频试验，以取得两个二进制的中位时刻。
trial (#) | `node` binary (ms) | modified `node` binary (ms)
=================================================================
1 | 56924 | 55011
2 | 52686 | 55869
3 | 59479 | 54043
4 | 54473 | 55229
5 | 52933 | 59723
=================================================================
average time: | 55299 | 55975两者都须要约莫一分钟来运转，因而基础没有太大差别，但让我们细致看看以确认我们的疑心是不是准确，我们运用Linux东西dtrace来评价V8渣滓网络器发作了什么。
GC（渣滓网络器）丈量时刻示意渣滓网络器完成单次扫描的完全周期的距离：
approx. time (ms) | GC (ms) | modified GC (ms)
=================================================
0 | 0 | 0
1 | 0 | 0
40 | 0 | 2
170 | 3 | 1
300 | 3 | 1
* * *
* * *
* * *
39000 | 6 | 26
42000 | 6 | 21
47000 | 5 | 32
50000 | 8 | 28
54000 | 6 | 35虽然这两个历程最先时雷同，但好像以雷同的速度运转GC，很明显，在恰当事情的背压体系几秒钟后，它将GC负载散布在4-8毫秒的一致距离内，直到数据传输完毕。
然则，当背压体系不到位时，V8渣滓网络最先迁延，一般二进制文件在一分钟内挪用GC约75次，但是，修正后的二进制文件仅触发36次。
这是因为内存运用量增添而积累的迟缓而渐进的债权，跟着数据传输，在没有背压体系的状况下，每一个块传输运用更多内存。
分派的内存越多，GC在一次扫描中须要处置惩罚的内存就越多，扫描越大，GC就越须要决议可以开释什么，而且在更大的内存空间中扫描星散的指针将斲丧更多的盘算才能。
内存耗尽
为肯定每一个二进制的内存斲丧，我们运用/usr/bin/time -lp sudo ./node ./backpressure-example/zlib.js零丁为每一个历程计时。
这是一般二进制的输出：
Respecting the return value of .write()
=============================================
real 58.88
user 56.79
sys 8.79
87810048 maximum resident set size
0 average shared memory size
0 average unshared data size
0 average unshared stack size
19427 page reclaims
3134 page faults
0 swaps
5 block input operations
194 block output operations
0 messages sent
0 messages received
1 signals received
12 voluntary context switches
666037 involuntary context switches虚拟内存占用的最大字节大小约为87.81mb。
如今变动.write()函数的返回值，我们获得：
Without respecting the return value of .write():
==================================================
real 54.48
user 53.15
sys 7.43
1524965376 maximum resident set size
0 average shared memory size
0 average unshared data size
0 average unshared stack size
373617 page reclaims
3139 page faults
0 swaps
18 block input operations
199 block output operations
0 messages sent
0 messages received
1 signals received
25 voluntary context switches
629566 involuntary context switches虚拟内存占用的最大字节大小约为1.52gb。
假如没有流来托付背压，则分派的内存空间要大一个数目级 — 统一历程之间的庞大差别！
这个试验展现了Node.js的反压机制是怎样优化和节约本钱的，如今，让我们理会一下它是怎样事情的！
背压怎样处理这些题目？
将数据从一个历程传输到另一个历程有差别的函数，在Node.js中，有一个名为.pipe()的内部内置函数，另有其他包也可以运用！但终究，在这个历程的基础层面，我们有两个自力的组件：数据泉源和斲丧者。
当从源挪用.pipe()时，它向斲丧者发出信号，示知有数据要传输，管道函数有助于为事宜触发器设置恰当的背压闭合。
在Node.js中，源是Readable流，而斲丧者是Writable流（这些都可以与Duplex或Transform流交换，但这超出了本指南的局限）。
触发背压的时刻可以准确地缩小到Writable的.write()函数的返回值，固然，该返回值由几个前提决议。
在数据缓冲区已凌驾highWaterMark或写入行列当前正忙的任何状况下，.write()将返回false。
当返回false值时，背压体系启动，它会停息传入的Readable流发送任何数据，并守候斲丧者再次预备就绪，清空数据缓冲区后，将发出.drain()事宜并恢复传入的数据流。
行列完成后，背压将许可再次发送数据，正在运用的内存空间将自行开释并为下一批数据做好预备。
这有效地许可在任何给定时刻为.pipe()函数运用牢固数目的内存，没有内存走漏，没有无穷缓冲，渣滓网络器只须要处置惩罚内存中的一个地区！
那末，假如背压云云主要，为何你（可以）没有听说过它？答案很简单：Node.js会自动为你完成一切这些事情。
那太好了！然则当我们试图相识怎样完成我们本身的自定义流时，也不是那末好。
注重：在大多数机械中，有一个字节大小可以肯定缓冲区什么时刻已满（在差别的机械上会有所差别），Node.js许可你设置本身的自定义highWaterMark，但平常，默认设置为16kb（16384，或objectMode流为16），在你可以愿望进步该值的状况下，可以尝试，然则要警惕！
.pipe()的生命周期
为了更好地明白背压，下面是一个关于Readable流的生命周期的流程图，该流被管道传输到Writable流中：
+===================+
x--> Piping functions +--> src.pipe(dest) |
x are set up during |===================|
x the .pipe method. | Event callbacks |
+===============+ x |-------------------|
| Your Data | x They exist outside | .on('close', cb) |
+=======+=======+ x the data flow, but | .on('data', cb) |
| x importantly attach | .on('drain', cb) |
| x events, and their | .on('unpipe', cb) |
+---------v---------+ x respective callbacks. | .on('error', cb) |
| Readable Stream +----+ | .on('finish', cb) |
+-^-------^-------^-+ | | .on('end', cb) |
^ | ^ | +-------------------+
| | | |
| ^ | |
^ ^ ^ | +-------------------+ +=================+
^ | ^ +----> Writable Stream +---------> .write(chunk) |
| | | +-------------------+ +=======+=========+
| | | |
| ^ | +------------------v---------+
^ | +-> if (!chunk) | Is this chunk too big? |
^ | | emit .end(); | Is the queue busy? |
| | +-> else +-------+----------------+---+
| ^ | emit .write(); | |
| ^ ^ +--v---+ +---v---+
| | ^-----------------------------------< No | | Yes |
^ | +------+ +---v---+
^ | |
| ^ emit .pause(); +=================+ |
| ^---------------^-----------------------+ return false; <-----+---+
| +=================+ |
| |
^ when queue is empty +============+ |
^------------^-----------------------< Buffering | |
| |============| |
+> emit .drain(); | ^Buffer^ | |
+> emit .resume(); +------------+ |
| ^Buffer^ | |
+------------+ add chunk to queue |
| <---^---------------------<
+============+注重：假如要设置管道以将一些流链接在一同来支配数据，则很可以会完成Transform流。
在这类状况下，你的Readable流的输出将输入到Transform中，并将管道到Writable中。
Readable.pipe(Transformable).pipe(Writable);背压将自动运用，但请注重，Transform流的输入和输出highWaterMark都可以被支配并将影响背压体系。
背压指南
从Node.js v0.10最先，Stream类供应了经由过程运用这些响应函数的下划线版原本修正.read()或.write()的行动的功用（._read()和._write()）。
关于完成Readable流和Writable流，有文档化的指南，我们假定你已浏览过这些内容，下一节将更深切一些。
完成自定义流时要恪守的划定规矩
流的黄金轨则始终是尊敬背压，最好实践的组成黑白抵牾的实践，只需你警惕避免与内部背压支撑相冲突的行动，你就可以肯定你遵照优越做法。
平常来讲：
- 假如你没有被请求，永久不要
.push()。 - 永久不要在返回
false后挪用.write()，而是守候’drain’。 - 流在差别的Node.js版本和你运用的库之间有变化，警惕并测试一下。
注重：关于第3点，构建浏览器流的异常有效的包是readable-stream，Rodd Vagg撰写了一篇很棒的博客文章，形貌了这个库的实用性，简而言之，它为Readable流供应了一种自动文雅降级，并支撑旧版本的浏览器和Node.js。
Readable流的特定划定规矩
到目前为止，我们已相识了.write()怎样影响背压，并将重点放在Writable流上，因为Node.js的功用，数据在技术上从Readable流向下流Writable。然则，正如我们可以在数据、物资或能量的任何传输中观察到的那样，源与目的一样主要，Readable流关于怎样处置惩罚背压至关主要。
这两个历程都相互依赖，有效地举行通讯，假如Readable疏忽Writable流请求它住手发送数据的时刻，那末.write()的返回值不准确就会有题目。
因而，关于.write()返回，我们还必需尊敬._read()要领中运用的.push()的返回值，假如.push()返回false值，则流将住手从源读取，不然，它将继承而不会停留。
以下是运用.push()的不好做法示例：
// This is problematic as it completely ignores return value from push
// which may be a signal for backpressure from the destination stream!
class MyReadable extends Readable {
_read(size) {
let chunk;
while (null !== (chunk = getNextChunk())) {
this.push(chunk);
}
}
}另外，在自定义流以外，存在疏忽背压的圈套，在这个优越的实践的反例中，运用顺序的代码会在数据可用时强迫经由过程（由.data事宜发出信号）：
// This ignores the backpressure mechanisms Node.js has set in place,
// and unconditionally pushes through data, regardless if the
// destination stream is ready for it or not.
readable.on('data', (data) =>
writable.write(data)
);Writable流的特定划定规矩
追念一下.write()可以会依据某些前提返回true或false，荣幸的是，在构建我们本身的Writable流时，流状态机将处置惩罚我们的回调并肯定什么时刻处置惩罚背压并为我们优化数据流。
然则，当我们想直接运用Writable时，我们必需尊敬.write()返回值并密切注重这些前提：
- 假如写行列忙，
.write()将返回false。 - 假如数据块太大，
.write()将返回false（该值由变量highWaterMark指导）。
// This writable is invalid because of the async nature of JavaScript callbacks.
// Without a return statement for each callback prior to the last,
// there is a great chance multiple callbacks will be called.
class MyWritable extends Writable {
_write(chunk, encoding, callback) {
if (chunk.toString().indexOf('a') >= 0)
callback();
else if (chunk.toString().indexOf('b') >= 0)
callback();
callback();
}
}
// The proper way to write this would be:
if (chunk.contains('a'))
return callback();
else if (chunk.contains('b'))
return callback();
callback();在完成._writev()时还须要注重一些事项，该函数与.cork()连系运用，但写入时有一个罕见毛病：
// Using .uncork() twice here makes two calls on the C++ layer, rendering the
// cork/uncork technique useless.
ws.cork();
ws.write('hello ');
ws.write('world ');
ws.uncork();
ws.cork();
ws.write('from ');
ws.write('Matteo');
ws.uncork();
// The correct way to write this is to utilize process.nextTick(), which fires
// on the next event loop.
ws.cork();
ws.write('hello ');
ws.write('world ');
process.nextTick(doUncork, ws);
ws.cork();
ws.write('from ');
ws.write('Matteo');
process.nextTick(doUncork, ws);
// as a global function
function doUncork(stream) {
stream.uncork();
}.cork()可以被挪用屡次，我们只须要警惕挪用.uncork()雷同的次数，使其再次活动。
结论
Streams是Node.js中常常运用的模块，它们关于内部结构异常主要，关于开发人员来讲，它们可以跨Node.js模块生态体系举行扩大和衔接。
愿望你如今可以举行毛病消除，平安地编写你本身的Writable和Readable流，并斟酌背压，并与同事和朋侪分享你的学问。
在运用Node.js构建运用顺序时，请务必浏览有关其他API函数的Stream的更多信息，以协助革新和开释你的流功用。
