在本章的开始，我本来只想写一些关于fs模块的内容，虽然这个模块包含的方法非常的多，但没有想到的是它和我们上一篇文章Node.js Buffer还存在着关联，所以我又把关于Buffer的内容简单的学习下，以至于对它不那么陌生。我以为这样就可以来看看fs模块了，没想到又杀出个程咬金，这就是本文要学习的Stream。
这个Stream目前还是个不稳定的模块，从API的介绍来说，所有stream是对象EventEmitter的实例，测试下：
console.log(new(require(‘stream’)) instanceof require(‘events’).EventEmitter);//输出：true
什么是Stream(流)
在Node中，Stream是一个抽象的接口，由不同的实例对象来实现，比如HTTP服务器中的requestI流，再比如stdout标准输出流。流可以是可读的，可写的，或者可读可写的。
为什么要提供Stream
在Node中，大部分的模块能持续的读或者写，为了确保这些模块拥有统一的方法来完成持续的读或者写，Stream于是为它们提供了统一的接口。Stream接口提供了通用的方法和属性，因为stream是EventEmitter的实例，所以允许这些模块emit自己的事件。
Readable Stream(可读流)
一个常见的readable stream就是从一个文件中读取数据。可读流支持的事件有四种：data，end,error,close。它们分别对应如下场景：
data：当有数据可读时触发
end：当没有数据可读时或者达到文件尾
error:读取数据时遇到异常
close：底层的文件描述(fd)关闭时
为了简单起见，我们使用fs模块中的方法来读取windows下的boot.ini文件：
|
1
2
3
4
5
6
|
var fs = require('fs');
fs.readFile('c:/boot.ini',function(err,data){
if(err) throw err;
//console.log(data);//传递给回调函数的data参数是一个Buffer对象，大小默认为64K?
console.log(data.toString());
}); |
fs.readFile方法能触发data事件，在这里我们把读取到的数据打印到控制台上。为了让本文稍微有点深度，这里贴上fs.readFile方法的源码：
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
fs.readFile = function(path, encoding_) {
var encoding = typeof(encoding_) === 'string' ? encoding_ : null;
var callback = arguments[arguments.length - 1];
if (typeof(callback) !== 'function') callback = function() {};
// first, stat the file, so we know the size.
var size;
var buffer; // single buffer with file data
var buffers; // list for when size is unknown
var pos = 0;
var fd;
fs.open(path, constants.O_RDONLY, 438 /*=0666*/, function(er, fd_) {
if (er) return callback(er);
fd = fd_;
fs.fstat(fd, function(er, st) {
if (er) return callback(er);
size = st.size;
if (size === 0) {
// the kernel lies about many files.
// Go ahead and try to read some bytes.
buffers = [];
return read();
}
buffer = new Buffer(size);
read();
});
});
function read() {
if (size === 0) {
buffer = new Buffer(8192);
fs.read(fd, buffer, 0, 8192, -1, afterRead);
} else {
fs.read(fd, buffer, pos, size - pos, -1, afterRead);
}
}
function afterRead(er, bytesRead) {
if (er) {
return fs.close(fd, function(er2) {
return callback(er);
});
}
if (bytesRead === 0) {
return close();
}
pos += bytesRead;
if (size !== 0) {
if (pos === size) close();
else read();
} else {
// unknown size, just read until we don't get bytes.
buffers.push(buffer.slice(0, bytesRead));
read();
}
}
function close() {
fs.close(fd, function(er) {
if (size === 0) {
// collected the data into the buffers list.
buffer = Buffer.concat(buffers, pos);
} else if (pos < size) {
buffer = buffer.slice(0, pos);
}
if (encoding) buffer = buffer.toString(encoding);
return callback(er, buffer);
});
}
}; |
想认真学习的人应该仔细读读fs.readFile是如何实现的，想看fs模块的完整源码位于lib目录下的fs.js文件。下载Node.js源码(版本v0.8.18)。
为了更加直观的描述readable stream中的这些事件，请参见下面的例子
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
var fs = require('fs');
var opt = {
flags:'r',
encoding:'utf8',
fd:null,
mode:0666,
bufferSize:64 * 1024,
start:0,
end:99
};st = fs.createReadStream('c:/boot.ini',opt);
//st.pause();//暂停data事件的触发//st.resume();//恢复data事件var text = "";
st.on('data',function(data){
text += data;
});st.on('end',function(close){
console.log("reading 99 bytes:\n" + text);
});st.on('error',function(error){
console.log('An error occurred:' + error);
});st.on('close',function(){
console.log('The file descriptor has been closed.');
});st.on('open',function(fd){
console.log('Current fd number: ' + fd);
}); |
输出：
Current fd number: 3
reading 99 bytes:
[boot loader]
timeout=3
default=multi(0)disk(0)rdisk(0)partition(1)\WINDOWS
[operating systems]
The file descriptor has been closed.
F:\wamp\www\Node>
不想太多解析这个示例，为什么文件描述是3？因为0，1，2已经被占用啦，原因？你懂的！
Writable Stream(可写流)
可写流支持的事件有：drain，error，close，pipe。我只介绍drain和pipe，它们分别对应如下场景：
drain：当Node的写队列为空并且Buffer中没有数据时，发生这个事件
pipe：当把Writable Stream对象传递一个Readable Stream对象的pipe方法时发生
分别来看看这两个事件的例子：
1.drain
|
01
02
03
04
05
06
07
08
09
10
11
|
var fs = require('fs');
var stream = fs.createWriteStream(__dirname + '/out.txt');
var str = 'i=[';
for (var i=0;i<10;i++){
stream.write(str + i + ']\r\n');
} stream.on('drain',function(){
console.log('start drain');
}); |
当for循环完成时，我们的控制台才打印出’start drain’，这意味着写队列已经为空，即当前没有事件存在写入的情况。由于drain事件能发生在任何时候，所以为了避免触发不必要的drain事件，官方的建议是，使用events.EventEmitter once()方法，这样drain事件只可能触发一次，尔后它上面的监听器就被删除了。
2.pipe
|
1
2
3
4
5
6
7
8
|
var fs = require('fs');
var writeStream = fs.createWriteStream('copyout.txt',{ flags:'w'});
var readStream = fs.createReadStream('./out.txt');
readStream.pipe(writeStream);writeStream.on('close',function(){
console.log('All done!');
}); |
这个示例把out.txt文件中的内容复制到了copyout.txt中。
这篇文章写了我很长时间，虽然还有很多的知识点没有介绍上来，也到此为止吧。我希望在日后有机会补充一些。
Date.prototype.format = function(format){
var o = {
"M+" : this.getMonth()+1, //month
"d+" : this.getDate(), //day
"h+" : this.getHours(), //hour
"m+" : this.getMinutes(), //minute
"s+" : this.getSeconds(), //second
"q+" : Math.floor((this.getMonth()+3)/3), //quarter
"S" : this.getMilliseconds() //millisecond
}
if(/(y+)/.test(format)) format=format.replace(RegExp.$1,
(this.getFullYear()+"").substr(4 - RegExp.$1.length));
for(var k in o)if(new RegExp("("+ k +")").test(format))
format = format.replace(RegExp.$1,
RegExp.$1.length==1 ? o[k] :
("00"+ o[k]).substr((""+ o[k]).length));
return format;
}
var fs=require('fs');
fs.open(__dirname+'/error.log','a',function(err,fd){
if(err){ throw err; }
cTime = new Date();
cTime = cTime.format('yyyy-MM-dd h:m:s');
var wBuff = new Buffer('Start log: ' + cTime + "\r\n");
buffPos = 0;
buffLen = wBuff.length;
filePos = 0;
fs.write(fd,wBuff,buffPos,buffLen,filePos,function(err,wbytes,data){
if(err){ throw err; }
console.log('wrote ' + wbytes + ' bytes');
fs.close(fd);
});
});