如何在 NodeJS 的自定义可读流中监听事件？答案

【问题标题】：How can I listen to an event inside a custom Readable stream in NodeJS?如何在 NodeJS 的自定义可读流中监听事件？
【发布时间】：2019-05-06 14:43:56
【问题描述】：

这是我的代码：

class Counter extends Readable {
    constructor() {
        super();
    }
    _read () { 
        readStream.on('data', (data) => {
            console.log(JSON.stringify(data));
            this.push(JSON.stringify(data));
        });
        readStream.on('end', () => {
            this.push(null);
        });
    }
}

我收到此错误，处理也变得非常缓慢。

(node:18488) MaxListenersExceededWarning: Possible EventEmitter memory 
leak detected. 11 end listeners added. Use emitter.setMaxListeners() 
to increase limit (node:18488) MaxListenersExceededWarning: Possible 
EventEmitter memory leak detected. 11 data listeners added. Use 
emitter.setMaxListeners() to increase limit

这是我的完整代码。基本上，我想阅读 CSV。将其转换为 JSON 并写入文件。我想使用流来执行此操作。我正在使用“csv-parser”npm 模块

const  fs = require('fs');
var Readable = require('stream').Readable; 
const csv = require('csv-parser');


class Counter extends Readable {
   constructor() {
       super();
   }
   _read () { 
       readStream.on('data', (data) => {
           console.log(JSON.stringify(data));
           this.push(JSON.stringify(data));
       });
       readStream.on('end', () => {
           this.push(null);
       });
   }
}
let writeStream = fs.createWriteStream('tmp1');
let readStream = fs.createReadStream('tmp.csv').pipe(csv());
var counter = new Counter(); 
counter.pipe(writeStream);

【问题讨论】：

标签： node.js events nodejs-stream readable

【解决方案1】：

我尝试将 1.48GB 的 csv 文件作为 json 数组写入文件，它没有给出您提到的任何错误或警告。

const http = require('http');
const fs = require('fs');
const csv = require('csv-parser');

const server = http.createServer();

const writeStr = fs.createWriteStream('data.json', { flags: 'a' });
writeStr.write('[\n');
let sep = "";

server.on('request', (req, res) => {
    console.log(req.method);

    fs.createReadStream('./Crimes_-_2001_to_present.csv').pipe(csv())
        .on('data', function (data) {
            writeStr.write(sep + JSON.stringify(data));
            if (!sep)
                sep = ",\n";
        })
        .on('end', function () {
            writeStr.write('\n]')
            console.log('DONE WRITING FILE');
            res.end();
        })
        .on('error', function (err) {
            consol.log(err);
        })
})
server.listen(8000);

仅供参考：数据集取自https://catalog.data.gov/dataset?res_format=CSV

【讨论】：