Node.js에서 Stream(읽기, 쓰기, 이중 및 변환 스트림)을 구현하는 방법에 대해 이야기해 보겠습니다.

const fs = require('fs'); const path = require('path'); const copy = (source, target) => { fs.readFile(path.resolve(source), (err, data) => { if(err) { throw new Error(err.toString()); return; } fs.writeFile(path.resolve(target), data, (err) => { if(!err) { console.log("复制成功！"); } }) }) }

const fs = require('fs'); function copy(source, target) { const rs = fs.createReadStream(source); const ws = fs.createWriteStream(target); rs.on('data', data => { ws.write(data); }); rs.on('end', () => { ws.end(); }); }

const fs = require('fs'); const rs = fs.createReadStream('a.txt', { highWaterMark: 1 // 缓存区最多存储 1 字节 }); rs.on('readable', () => { let data; while(data=rs.read()) { console.log(data.toString()); } })

const fs = require('fs'); const rs = fs.createReadStream('a.txt', { highWaterMark: 2 }); rs.on('data', data => { console.log(data.toString()); }); rs.on('end', () => { console.log("文件读取完毕！"); });

const http = require('http'); const app = http.createServer(); app.on('request', (req, res) => { let datas = []; req.on('data', data => { datas.push(data); }); req.on('end', () => { req.body = Buffer.concat(datas); // 当读取完 body 中的内容之后，将内容返回给客户端 res.end(req.body); }); }) app.listen(3000, () => { console.log("服务启动在 3000 端口... ..."); })

const fs = require('fs'); const ws = fs.createWriteStream('b.txt', { highWaterMark: 16 * 1024 }); function writeMillionTimes(writer, data, encoding, callback) { let i = 10000; write(); function write() { // 表示是否可以向可写流中写入数据 let ok = true; while(i-- > 0 && ok) { // 当 writer.write() 方法返回 false 表示不可写入数据 ok = writer.write(data, encoding, i === 0 ? callback : null); } if(i > 0) { // 说明 ok 为 false，即不能向缓冲区中写入内容了 console.log("drain", i); // 监听 drain 事件，当队列消费完毕时继续调用 write() 方法写入 writer.once('drain', write); } } } writeMillionTimes(ws, 'simple', 'utf-8', () => { console.log("end"); })

drain 7268 drain 4536 drain 1804 end

const fs = require('fs'); const ws = fs.createWriteStream('b.txt'); ws.write('Hello'); ws.write('World'); ws.end('!'); ws.on('close', () => { console.log("close"); // close })

const fs = require('fs'); const ws = fs.createWriteStream('b.txt'); ws.write('Hello'); ws.write('World'); ws.end('!'); ws.write('write again'); // Error [ERR_STREAM_WRITE_AFTER_END]: write after end

const fs = require('fs'); const ws = fs.createWriteStream('b.txt'); ws.write('Hello'); ws.write('World'); ws.end('!'); ws.on('close', () => { console.log("close"); }); ws.on('finish', () => { console.log("finish"); });

finish close

const http = require('http'); const fs = require('fs'); const app = http.createServer(); app.on('request', (req, res) => { const rs = fs.createReadStream('index.html'); rs.on('data', data => { res.write(data); }) rs.on('end', () => { res.end() }); }); app.listen(3000, () => { console.log("服务启动在 3000 端口 ... ..."); })

const fs = require('fs'); function copy(source, target) { const rs = fs.createReadStream(source); const ws = fs.createWriteStream(target); rs.on('data', data => { ws.write(data); }); rs.on('end', () => { ws.end(); }); } copy('a.txt', 'b.txt');

const fs = require('fs'); function copy(source, target) { const rs = fs.createReadStream(source); const ws = fs.createWriteStream(target); rs.on('data', data => { if (!ws.write(data)) { rs.pause(); } }); rs.on('end', () => { ws.end(); }); ws.on('drain', () => { rs.resume(); }) }

const fs = require('fs'); function copy(source, target) { const rs = fs.createReadStream(source); const ws = fs.createWriteStream(target); rs.pipe(ws); }

Readable.prototype.pipe = function(ws) { this.on('data', data => { if (!ws.write(data)) { this.pause(); } }); ws.on('drain', () => { this.resume(); }); // 触发 pipe 事件 ws.emit('pipe', this); // 返回可写流，以支持链式调用 return ws; }

+===================+ 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 | | <---^---------------------< +============+

const { Readable } = require('stream'); class IeteratorReadableStream extends Readable { constructor(iterator) { super(); this.iterator = iterator; } _read() { let data = this.iterator.next(); // console.log(data); if(data.done) { this.push(null); } else { // 必须 push 字符串或者 Buffer this.push(data.value+''); } } } module.exports = IeteratorReadableStream;

const IeteratorReadableStream = require('./IteratorReadableStream'); function *getData() { for(let i = 0; i < 5; i++) { yield i; } } let rs = new IeteratorReadableStream(getData()); rs.on('data', data => { console.log(data.toString()); }); rs.on('end', () => { console.log("迭代结束"); });

0 1 2 3 4 迭代结束

const fs = require('fs'); const { Writable } = require('stream'); class FileWritableStream extends Writable { constructor(filepath) { super(); this.filepath = filepath; } _write(chunk, encoding, callback) { fs.appendFile(this.filepath, chunk, { encoding }, callback) } }

process.stdin.pipe(new FileWritableStream('c.txt'));

const { Duplex } = require('stream'); class CustomDuplexStream extends Duplex { constructor() { super(); this.currentCharCode = 65; } _read() { if(this.currentCharCode <= 90) { this.push(String.fromCharCode(this.currentCharCode++)) } else { this.push(null); } } _write(chunk, encoding, callback) { console.log(chunk.toString()); callback(); } }

const dp = new CustomDuplexStream(); dp.write("1"); dp.write("2"); dp.end(); dp.pipe(process.stdout);

1 2 ABCDEFGHIJKLMNOPQRSTUVWXYZ

const { Transform } = require('stream'); class UpperTransformStream extends Transform { _transform(chunk, encoding, callback) { this.push(chunk.toString().toUpperCase()); callback(); } }

const ts = new UpperTransformStream(); const rs = fs.createReadStream('a.txt'); rs.pipe(ts).pipe(process.stdout);

const zlib = require('zlib'); const fs = require('fs'); const args = process.argv.slice(2); const source = fs.createReadStream(args[0]); const target = fs.createWriteStream(args[1]); const gzip = zlib.createGzip(); source.pipe(gzip).pipe(target);

node gzip.js Graph.md Graph.md.gz