<code>var Readable = require('stream').Readable; var rs = Readable(); var c = 97 - 1; rs._read = function () { if (c >= 'z'.charCodeAt(0)) return rs.push(null); setTimeout(function () { rs.push(String.fromCharCode(++c)); }, 100); }; rs.pipe(process.stdout); process.on('exit', function () { console.error('\n_read() called ' + (c - 97) + ' times'); }); process.stdout.on('error', process.exit);</code>
用下面的命令运行程序我们发现_read()方法只调用了5次：
<code>$ node read2.js | head -c5 abcde _read() called 5 times</code>
使用计时器的原因是系统需要时间来发送信号来通知程序关闭管道。使用process.stdout.on('error', fn) 是为了处理系统因为header命令关闭管道而发送SIGPIPE信号，因为这样会导致process.stdout触发EPIPE事件。如果想创建一个的可以压入任意形式数据的可读流，只要在创建流的时候设置参数objectMode为true即可，例如：Readable({ objectMode: true })。
2>读取readable stream数据
大部分情况下我们只要简单的使用pipe方法将可读流的数据重定向到另外形式的流，但是在某些情况下也许直接从可读流中读取数据更有用。如下：
<code>process.stdin.on('readable', function () { var buf = process.stdin.read(); console.dir(buf); }); $ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume0.js <buffer 0a="" 61="" 62="" 63=""> <buffer 0a="" 64="" 65="" 66=""> <buffer 0a="" 67="" 68="" 69=""> null</buffer></buffer></buffer></code>
当可读流中有数据可读取时，流会触发'readable' 事件，这样就可以调用.read()方法来读取相关数据，当可读流中没有数据可读取时，.read() 会返回null，这样就可以结束.read() 的调用， 等待下一次'readable' 事件的触发。下面是一个使用.read(n)从标准输入每次读取3个字节的例子：
<code>process.stdin.on('readable', function () { var buf = process.stdin.read(3); console.dir(buf); });</code>
如下运行程序发现，输出结果并不完全!
<code>$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume1.js <buffer 61="" 62="" 63=""> <buffer 0a="" 64="" 65=""> <buffer 0a="" 66="" 67=""></buffer></buffer></buffer></code>
这是应为额外的数据数据留在流的内部缓冲区里了，而我们需要通知流我们要读取更多的数据.read(0)可以达到这个目的。
<code>process.stdin.on('readable', function () { var buf = process.stdin.read(3); console.dir(buf); process.stdin.read(0); });</code>
这次运行结果如下：
<code>$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume2.js <buffer 0a="" 64="" 65=""> <buffer 0a="" 68="" 69=""></buffer></buffer></code>
我们可以使用 .unshift() 将数据重新押回流数据队列的头部，这样可以接续读取押回的数据。如下面的代码，会按行输出标准输入的内容：
<code>var offset = 0; process.stdin.on('readable', function () { var buf = process.stdin.read(); if (!buf) return; for (; offset < buf.length; offset++) { if (buf[offset] === 0x0a) { console.dir(buf.slice(0, offset).toString()); buf = buf.slice(offset + 1); offset = 0; process.stdin.unshift(buf); return; } } process.stdin.unshift(buf); }); $ tail -n +50000 /usr/share/dict/american-english | head -n10 | node lines.js 'hearties' 'heartiest' 'heartily' 'heartiness' 'heartiness\'s' 'heartland' 'heartland\'s' 'heartlands' 'heartless' 'heartlessly'</code>
当然，有很多模块可以实现这个功能，如：split 。
3-3、writable streams
writable streams只可以作为.pipe()函数的目的参数。如下代码：
<code>src.pipe( writableStream );</code>
1>创建 writable stream
重写 ._write(chunk, enc, next) 方法就可以接受一个readable stream的数据。
<code>var Writable = require('stream').Writable; var ws = Writable(); ws._write = function (chunk, enc, next) { console.dir(chunk); next(); }; process.stdin.pipe(ws); $ (echo beep; sleep 1; echo boop) | node write0.js <buffer 0a="" 62="" 65="" 70=""> <buffer 0a="" 62="" 6f="" 70=""></buffer></buffer></code>
第一个参数chunk是数据输入者写入的数据。第二个参数end是数据的编码格式。第三个参数next(err)通过回调函数通知数据写入者可以写入更多的时间。如果readable stream写入的是字符串，那么字符串会默认转换为Buffer，如果在创建流的时候设置Writable({ decodeStrings: false })参数，那么不会做转换。如果readable stream写入的数据时对象，那么需要这样创建writable stream
<code>Writable({ objectMode: true })</code>
2>写数据到 writable stream
调用writable stream的.write(data)方法即可完成数据写入。
<code>process.stdout.write('beep boop\n');</code>
调用.end()方法通知writable stream 数据已经写入完成。