Nodejs流学习系列之二: Writable Stream

前言

上一篇文章Nodejs流学习系列之一: Readable Stream我们简单地介绍了Nodejs的流的概念以及分类.然后详细地解释了可读流的两种模式以及缓存,简单地回顾了可读流的API.这一篇我们将介绍可写流,基本概念上一篇已经提过,就不再赘述,直奔主题吧~

1 Writable流的基本形态

我们使用最最基本的语法新建一个可写流,代码如下:

1class MyWritable extends Writable {
2  constructor(options) {
3    // Calls the stream.Readable(options) constructor
4    super(options);
5  }
6  _write(chunk, encoding, callback) {
7  }
8}
9
10const ws = new MyWritable()

这个实例化后的Writable流,其内部结构是这样的:

1MyWritable {
2  _writableState:
3   WritableState {
4     objectMode: false, // 是否是object mode
5     highWaterMark: 16384, // 最高水位,默认是16K
6     needDrain: false, // 是否需要发送drain事件
7     ending: false, // 是否可写流正在关闭,在调用end()事件之前
8     ended: false, // 可写流是否已经停止
9     finished: false, // 标识已经发送过finish事件
10     decodeStrings: true, // 标识是否我们应该在传递给_write之前编码string为buffer
11     defaultEncoding: 'utf8', // 默认编码
12     length: 0, // 该字段标识的是等待push到底层socket或者文件的数据大小
13     writing: false, // 标识是否正在写缓存到底层资源中
14     corked: 0, // 标识是否需要先缓存直到调用uncork()方法才会写到底层资源
15     sync: true, // 标识onwrite的cb调用是立即调用还是下一个tick调用
16     bufferProcessing: false, // 标识我们是否正在处理之前缓存的items
17     onwrite: [Function: bound onwrite], // 这个回调会传递给_write(chunk, cb)的cb参数
18     writecb: null, // 用户提供给write(chunk,encoding,cb)的回调
19     writelen: 0, // 标识当_write被调用的时候需要写的数据的大小
20     bufferedRequest: null, // 等待需要写到底层资源的缓存请求,是一个"链表"
21     lastBufferedRequest: null, // 缓存请求的最后一个
22     pendingcb: 0, // 用户当前待处理的write回调,该值在finish事件发射之前必须为0
23     prefinished: false, // 如果我们等待的唯一一件事就是_write cbs的话就发射`prefinish`事件,该事件供同// 步Transform流使用
24     errorEmitted: false, // 标识我们已经发射过错误事件,并且不应该再发射了
25     bufferedRequestCount: 0, // 缓存请求的个数
26     corkedRequestsFree: // 分配第一个CorkedRequest: {next:null,entry:null,finish:undefined }
27      CorkedRequest {
28        next: null,
29        entry: null,
30        finish: [Function: bound onCorkedFinish] } },
31  writable: true,
32  domain: null,
33  _events: {},
34  _eventsCount: 0,
35  _maxListeners: undefined }

内部结构的每个字段的含义都已经标明,这个时候你可以将其和可读流进行对比,在缓存这块的组织还是基本一样的.

有了内部结构我们再贴上一张可写流的原理：

2 可写流的原理

在讲原理之前我们完善一下上面的demo,如下:

1const { Writable } = require('stream')
2
3class MyWritable extends Writable {
4  constructor(options) {
5    // Calls the stream.Readable(options) constructor
6    super(options);
7  }
8  _write(chunk, encoding, callback) {
9    console.log('we write--', chunk)
10    // callback()
11  }
12}
13
14const ws = new MyWritable()
15
16console.log('writable stream: ', ws)
17
18const ws1 = ws.write('abcdefgh')
19
20console.log('writable stream: ', ws)
21console.log('write buffer return value:', ws1)
22console.log(ws._writableState.getBuffer())
23
24const ws2 = ws.write('ijk')
25
26console.log('writable stream: ', ws)
27console.log('write buffer return value:', ws2)
28console.log(ws._writableState.getBuffer())
29
30const ws3 = ws.write('opq')
31
32console.log('writable stream: ', ws)
33console.log('write buffer return value:', ws3)
34console.log(ws._writableState.getBuffer())

请关注我们每次写值进去对应的可写流实例的状态变化以及buffer的变化.

结果如下:(结果有点长,为了能够直观看出来就不要介意了~~)

1writable stream:  MyWritable {
2  _writableState:
3   WritableState {
4     objectMode: false,
5     highWaterMark: 16384,
6     needDrain: false,
7     ending: false,
8     ended: false,
9     finished: false,
10     decodeStrings: true,
11     defaultEncoding: 'utf8',
12     length: 0,
13     writing: false,
14     corked: 0,
15     sync: true,
16     bufferProcessing: false,
17     onwrite: [Function: bound onwrite],
18     writecb: null,
19     writelen: 0,
20     bufferedRequest: null,
21     lastBufferedRequest: null,
22     pendingcb: 0,
23     prefinished: false,
24     errorEmitted: false,
25     bufferedRequestCount: 0,
26     corkedRequestsFree:
27      CorkedRequest {
28        next: null,
29        entry: null,
30        finish: [Function: bound onCorkedFinish] } },
31  writable: true,
32  domain: null,
33  _events: {},
34  _eventsCount: 0,
35  _maxListeners: undefined }
36we write-- <Buffer 61 62 63 64 65 66 67 68>
37writable stream:  MyWritable {
38  _writableState:
39   WritableState {
40     objectMode: false,
41     highWaterMark: 16384,
42     needDrain: false,
43     ending: false,
44     ended: false,
45     finished: false,
46     decodeStrings: true,
47     defaultEncoding: 'utf8',
48     length: 8,
49     writing: true,
50     corked: 0,
51     sync: false,
52     bufferProcessing: false,
53     onwrite: [Function: bound onwrite],
54     writecb: [Function: nop],
55     writelen: 8,
56     bufferedRequest: null,
57     lastBufferedRequest: null,
58     pendingcb: 1,
59     prefinished: false,
60     errorEmitted: false,
61     bufferedRequestCount: 0,
62     corkedRequestsFree:
63      CorkedRequest {
64        next: null,
65        entry: null,
66        finish: [Function: bound onCorkedFinish] } },
67  writable: true,
68  domain: null,
69  _events: {},
70  _eventsCount: 0,
71  _maxListeners: undefined }
72write buffer return value: true
73[]
74writable stream:  MyWritable {
75  _writableState:
76   WritableState {
77     objectMode: false,
78     highWaterMark: 16384,
79     needDrain: false,
80     ending: false,
81     ended: false,
82     finished: false,
83     decodeStrings: true,
84     defaultEncoding: 'utf8',
85     length: 11,
86     writing: true,
87     corked: 0,
88     sync: false,
89     bufferProcessing: false,
90     onwrite: [Function: bound onwrite],
91     writecb: [Function: nop],
92     writelen: 8,
93     bufferedRequest:
94      WriteReq {
95        chunk: <Buffer 69 6a 6b>,
96        encoding: 'buffer',
97        callback: [Function: nop],
98        next: null },
99     lastBufferedRequest:
100      WriteReq {
101        chunk: <Buffer 69 6a 6b>,
102        encoding: 'buffer',
103        callback: [Function: nop],
104        next: null },
105     pendingcb: 2,
106     prefinished: false,
107     errorEmitted: false,
108     bufferedRequestCount: 1,
109     corkedRequestsFree:
110      CorkedRequest {
111        next: null,
112        entry: null,
113        finish: [Function: bound onCorkedFinish] } },
114  writable: true,
115  domain: null,
116  _events: {},
117  _eventsCount: 0,
118  _maxListeners: undefined }
119write buffer return value: true
120[ WriteReq {
121    chunk: <Buffer 69 6a 6b>,
122    encoding: 'buffer',
123    callback: [Function: nop],
124    next: null } ]
125writable stream:  MyWritable {
126  _writableState:
127   WritableState {
128     objectMode: false,
129     highWaterMark: 16384,
130     needDrain: false,
131     ending: false,
132     ended: false,
133     finished: false,
134     decodeStrings: true,
135     defaultEncoding: 'utf8',
136     length: 14,
137     writing: true,
138     corked: 0,
139     sync: false,
140     bufferProcessing: false,
141     onwrite: [Function: bound onwrite],
142     writecb: [Function: nop],
143     writelen: 8,
144     bufferedRequest:
145      WriteReq {
146        chunk: <Buffer 69 6a 6b>,
147        encoding: 'buffer',
148        callback: [Function: nop],
149        next: [Object] },
150     lastBufferedRequest:
151      WriteReq {
152        chunk: <Buffer 6f 70 71>,
153        encoding: 'buffer',
154        callback: [Function: nop],
155        next: null },
156     pendingcb: 3,
157     prefinished: false,
158     errorEmitted: false,
159     bufferedRequestCount: 2,
160     corkedRequestsFree:
161      CorkedRequest {
162        next: null,
163        entry: null,
164        finish: [Function: bound onCorkedFinish] } },
165  writable: true,
166  domain: null,
167  _events: {},
168  _eventsCount: 0,
169  _maxListeners: undefined }
170write buffer return value: true
171[ WriteReq {
172    chunk: <Buffer 69 6a 6b>,
173    encoding: 'buffer',
174    callback: [Function: nop],
175    next:
176     WriteReq {
177       chunk: <Buffer 6f 70 71>,
178       encoding: 'buffer',
179       callback: [Function: nop],
180       next: null } },
181  WriteReq {
182    chunk: <Buffer 6f 70 71>,
183    encoding: 'buffer',
184    callback: [Function: nop],
185    next: null } ]
186

2.1 可写流状态的变化

1. 第一次打印可读流,大部分的字段都是false或者null
2. 写入'abcdefgh',这个时候可读流的状态开始发生变化: 2.1 触发了_write方法,打印了写入的数据 2.2 length/writelen字段都变为8,并且writing=true,pendingcb=1 2.3 write方法返回仍为true 2.4 getBuffer返回的数组为空[]
3. 写入'ijk',这个时候可写流的状态继续变化: 3.1 这次没有触发_write方法 3.2 length字段变为11,pendingcb=2 3.3 bufferedRequest/lastBufferedRequest有数据了,结构为{chunk,encoding,callback,next},其中next为null 3.4 bufferedRequestCount=1 3.5 write方法返回仍为true,但是getBuffer返回的不再是空数组
4. 写入'opq',这个时候的可写流状态有规律可循,基本是在第三个步骤的基础上累加: 3.1 这次没有触发_write方法 3.2 length字段变为14,pendingcb=3 3.3 bufferedRequest/lastBufferedRequest有数据了,结构为{chunk,encoding,callback,next},并且next有数据了 3.4 bufferedRequestCount=2 3.5 write方法返回仍为true,但是getBuffer返回的不再是空数组

这个时候请问以下问题:

1. 为什么不再调用_write方法了?
2. 第一次写入的时候为什么buffer仍然为null?
3. write方法什么时候会返回false?
4. writelen为什么一直是8?

想要解答上面的问题,我们需要从源码去解析可写流实现的机制.下图是我画的一个大致的执行流程图

从上面的代码执行流程我们可以清楚地解释上面的四个问题：

1. 因为在_write方法中没有调用callback，导致没有调用代码中state.onwrite方法，进而不会更新状态，使writing一直是true
2. 因为第一次是直接调用doWrite方法写到底层资源上，并不会去走缓存的那个判断
3. 当state.length < state.highWaterMark不成立的时候就回返回true
4. 这个也是说明writelen是指的是等待写入底层资源的数据长度。因为没有再触发任何的_write，所以这个值一直是第一次写入的数据的长度

至此我们应该也大致明白了可写流的一些基本逻辑，我们接下去看看buffer的组织。

2.2 buffer的组织

可写流的buffer组织和可读流差不多，都是通过类似队列的形式去组织，形如：

1bufferedRequest:
2WriteReq {
3    chunk: <Buffer 69 6a 6b>,
4    encoding: 'buffer',
5    callback: [Function: nop],
6    next: [Object] },
7lastBufferedRequest:
8  WriteReq {
9    chunk: <Buffer 6f 70 71>,
10    encoding: 'buffer',
11    callback: [Function: nop],
12    next: null },

即{chunk: data, next: { chunk: data, next: .....}},然后比可读流多了一个回调函数(callback)和编码(encoding)。组织该buffer的代码如下：

1var last = state.lastBufferedRequest;
2state.lastBufferedRequest = {
3  chunk,
4  encoding,
5  isBuf,
6  callback: cb,
7  next: null
8};
9if (last) {
10  last.next = state.lastBufferedRequest;
11} else {
12  state.bufferedRequest = state.lastBufferedRequest;
13}
14state.bufferedRequestCount += 1;

3. 可写流的事件介绍

可写流实现了close/error/drain/finish/pipe/unpipe六种事件.比较生疏的事件是drain。

3.1 Drain事件

当你调用stream.write方法的时候返回false,这个时候表明写入的数据超过可写流的阀值.然后等待可写流又能恢复写入数据的时候,该事件将会被出发.

在源码中我们发现有这么一个变量: this.needDrain = false;,该变量标识是否需要发送drain事件. 该变量在下面的条件下会置为true:

1function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) {
2  ...
3  // we must ensure that previous needDrain will not be reset to false.
4  if (!ret)
5    state.needDrain = true;
6  ...
7}

之后再等到buffer又重新清空的时候将其置为false:

1function onwriteDrain(stream, state) {
2  if (state.length === 0 && state.needDrain) {
3    state.needDrain = false;
4    stream.emit('drain');
5  }
6}

3.2 其他事件

可写流的写入完成事件是finish,这个与可读流的读完事件end一样,只是名字有所区别而已. 而pipe事件和unpipe事件则是当有可读流将此可写流作为输出的时候会触发此事件.

4。 可写流的水位

水位的判断在writeOrBuffer中判断:

1  var ret = state.length < state.highWaterMark;
2  // we must ensure that previous needDrain will not be reset to false.
3  if (!ret)
4    state.needDrain = true;

还记得在上一篇文章说的，可写流的水位只是做为write方法是否返回false的依据，它并不会暂停写数据到缓存中去，所以数据还是会继续保存。因此在代码中你并没有看到直接return的代码。

5. 可写流的方法

可写流的方法有cork/uncork/end/setDefaultEncoding/write/destroy

5.1 cork()/uncork()方法

cork/uncork是组合使用的,前者可以直接强制数据滞留到缓存中,然后当我们再调用uncork或者end的时候再写到具体的底层资源.

这种情况是为了避免我们频繁地写小块的数据到可写流中造成性能的下降,所以我们可以批量写完一批数据后再调用uncork事件来flush数据.在官方文档中建议uncork放在process.nextTick()中延迟调用.

5.2 write(chunk[, encoding][, callback])

该方法用来写指定的数据到可写流中,如果返回true表明内部可用的缓存仍然小于highWaterMark。如果为true的话那就是内部缓存空间不够,此时写入数据就应该暂停直到接收到drain事件。但是并不表示此时你再往可写流写数据会丢数据,Nodejs仍然会缓存所有写入的chunks直到使用完所有的缓存,这个时候将会无条件地终止掉缓存数据.甚至在中断之前,高内存的使用将会导致低效的垃圾回收性能以及高RSS(指的是没有释放内存回系统即使这块内存不再使用).

当可写流没有耗干之前就写数据对于Transform流来说是一个特殊的问题,因为Transform流是默认暂停的直到他有管道输出流或者有data/readable事件处理的时候.

5.3 其他方法

1. end([chunk][, encoding][, callback])方法： 该方法是用来告知可写流已经没有别的数据可写入了。可选项chunk和encoding允许在最后关闭可写流之前再直接写入最后的编码数据。
2. setDefaultEncoding(encoding)方法： 与可读流的setDefaultEncoding(encoding)方法类似
3. destroy([error])方法： 与可读流的destroy([error])方法类似

6. 可写流实例需要实现的方法

可读流实例只需要实现一个_read方法，但是可写流可以最多实现三个方法： _write, _writev, _final。我们一一介绍这三个方法都是需要做些什么。

6.1 _write(chunk, encoding, callback)

该方法用来将数据写到底层资源中。其中callback是必须调用以通知写入成功或者出现错误。如果不调用，在之前的例子我们也已经看到其后果了。

如果decodeStrings为true的话，那么chunk也许是一个字符串而不是一个Buffer，encoding将会标识字符串的字符编码。如果为false的话，那么encoding参数可以安全地忽略掉，那么chunk将保留相同的对象。

如果我们想要处理chunk数组的话，那我们需要实现下面的这个方法。

6.2 _writev(chunks, callback)

通过参数我们可以看到该方法是用于一次性处理多个chunk，chunks是一个数组。如果该方法有实现，那么我们会优先使用该方法去实现写缓存到底层资源，因为这是一种更快地方法，具体代码如下：

1function clearBuffer(stream, state) {
2  ...
3  if (stream._writev && entry && entry.next) {
4    // Fast case, write everything using _writev()
5    ...
6  } else {
7    // Slow case, write chunks one-by-one
8    ...
9  }
10}

6.3 _final(callback)

该方法可以在流关闭之前调用。延迟finish事件直到callback方法被调用。这个在我们想要关闭流之前关闭资源或者写入额外数据的时候比较有用。

参考

1. nodejs Stream