深入理解并运用Node中的IO模型流
Posted Failarmy
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了深入理解并运用Node中的IO模型流相关的知识，希望对你有一定的参考价值。
在 NodeJs 中，流随处可见，读/写文件流，HTTP请求/返回流，stdin/stdout流。理解并运用好流会让你的Node更具力量。
Stream
lib/_stream_readable.js
lib/_stream_writable.js
lib/_stream_tranform.js
lib/_stream_duplex.js
流主要有可读 Readable，可写流 Writable，双工可读可写流 Duplex， Transform 流就是继承 Duplex 的。 通过pipe管道，可读流可以pipe到一个或多个可写流。 看源码能发现里面涉及了一堆状态控制的代码，什么时候读，什么时候写，什么时候暂停读。 大部分情况下程序面对的问题。通常都可以抽象成一个输入/输出的问题，中间可能会包含转换。
实际问题怎么运用流呢。
读取大文件
小文件可以一次性读到内存，但如果一个 10G 的文件呢？ReadLine 模块很好用但是你知道背后怎么实现的吗。 试试怎么用 stream.Transform 来自己实现一个readLine
const stream = require("stream");
const fs = require("fs");
function ReadLineStream(FILE_PATH) {
var transformer = new stream.Transform({ objectMode: true });
// 实现Transform主要的_transform方法
transformer._transform = function(chunk, encoding, done) {
// buffer 需要toString
var data = chunk.toString();
// 累加读取出来的数据
if (this._lastLineData) data = this._lastLineData + data;
// 按照换行符分割
var lines = data.split("\n");
// 取出数组末尾剩余不是一行的内容和下一次拼接
this._lastLineData = lines.splice(lines.length - 1, 1)[0];
// 通过this.push 对外输出剩余累计的行
lines.forEach(this.push.bind(this));
done();
};
transformer._flush = function(done) {
// 末尾处理
if (this._lastLineData) this.push(this._lastLineData);
this._lastLineData = null;
done();
};
// 创建文件读取流
var source = fs.createReadStream(FILE_PATH);
source.pipe(transformer);
return transformer;
}
var fileReader = ReadLineStream("xx.log");
// 创建一个可写流
var writable = Stream.Writable({
objectMode: true,
write: function(line, _, next) {
async () => {
// 逐行写入到mysql
var parsed = JSON.parse(line);
await mysql.insert("insert into log set ?", parsed);
process.nextTick(next);
};
return true;
}
});
fileReader.pipe(writable);
通过实现一个消费/可写流我们就可以来对大文件进行处理，比如说实现一个 word count 计数器， 从文件导入到数据库。
而不用担心需要一次性读取整个文件到内存里out of memory这种问题。
通过继承 strean 模块我们也不需要过多的去考虑什么时候该读，什么时候因为写的压力大（背压），该停止读，让整个读写流有序的运行。 你只需要专注于实现你自己的 write read transform Spark, Strom 的实时计算流也是这样的，大任务分解成小任务，只需要专注于自己业务逻辑的 map，reduce
单机爬虫
再举个栗子, 它的输入可能是一堆 URL、输出是结构化的数据。需要写入到关系型数据库。
可以把 URL 数据获取抽象成一个可读流，
爬取过程，数据提取抽象成一个 transform 流
写入数据库抽象成一个可写流，
只需要约定好每个过程输出的数据模型，就可以在每个过程实现各种目的不一样的流。
如数据源，可以是读取文件，MYQL，分布式列队
抓取转换流，可以是普通的 HTTP爬虫，Puppeteer 可渲染性爬虫
数据存储流，可以写文件，MYQL，或者HDFS
read > transform > write
这样程序看起来是不是特别简洁？
URL 读取流
const Stream = require('stream');
const fs = require('fs');
const Sequelize = require('sequelize');
const Op = Sequelize.Op
const sequelize = new Sequelize('database', 'username', 'password', {
host: 'localhost',
dialect: 'mysql',
pool: {
max: 5,
min: 0,
acquire: 30000,
idle: 10000
},
});
// mysql表模型
const Url = sequelize.define('urls', {
id: {
type: Sequelize.INTEGER,
primaryKey: true,
autoIncrement: true,
},
url: Sequelize.STRING,
});
class SpiderMysqlSourceReadStream extends Stream.Readable {
constructor(opts){
super();
this.opts = opts || {};
this.filePath = opts.file || '';
this.fileStream = null;
this.cursor = 0;
this.connection = null;
Stream.Readable.call(this, {
objectMode: true,
highWaterMark: opts.highWaterMark || 1000
});
}
// 覆盖并实现read方法
_read(){
// 查大于当前id的10条
var urls = await Model.findAll({
attributes: ['url', 'id'],
where: {
id: {
[Op.gt]: this.cursor
}
},
limit: 10
});
urls.forEach((row) => {
this.cursor = row.id;
this.push(row);
});
}
}
把 URL 的变成结构化数据的转换流
const cheerio = require('cheerio')
class SpiderFetch extends Stream.Transform {
_tranform(row, encoding, done){
var url = row.url;
(async () => {
var parsed = null;
try{
var contentResp = await fetch(url);
var contexthtml = await contentResp.text();
parsed = this.parse(contextHtml);
}catch(e){
console.log('fetch error', e)
}
this.push(parsed);
done();
})();
}
parse(contextHtml){
var $ = cheerio.load(contextHtml);
....
....
}
}
mysql 入库写入流
// mysql表模型
const Document = sequelize.define('document', {
id: {
type: Sequelize.INTEGER,
primaryKey: true,
autoIncrement: true,
},
title: Sequelize.STRING,
});
const MysqlWriteStream extends Stream.Writable {
_write(chunk, encoding, done){
var data = JSON.parse(chunk.toString());
(async () => {
await Document.create(data);
done();
})();
}
}
var readStream = new SpiderMysqlSourceReadStream();
var fetchStream = new SpiderFetch();
var writeStream = new MysqlWriteStream();
readStream.pipe(fetchStream).pipe(writeStream);
是不是简洁明了？ 在Node中异步流随处可见，谁让它基因就是这样呢。
参考：
https://nodejs.org/api/stream.html
https://www.barretlee.com/blog/2017/06/06/dive-to-nodejs-at-stream-module/
https://nodejs.org/en/docs/guides/backpressuring-in-streams/
以上是关于深入理解并运用Node中的IO模型流的主要内容，如果未能解决你的问题，请参考以下文章