Node JS 关闭一个读取流

Question

我坚持关闭读取流。我正在使用 csv-parser 模块从 CSV 文件中读取数据，进行一些处理并将数据写入 MongoDB。一切正常，除了我无法退出我的程序。它只是等待，我必须强制退出它。我怎样才能完成它的执行？

const main = () => {

  const records = [];

  fs.readdir(dataPath, (err, files) => {

    if (err) console.log("Failed to read. ", err);

    else {

      fs.createReadStream(`${dataPath}/${files[0]}`)

        .pipe(csv({ skipLines: 7, mapHeaders: ({ header, index }) => _.camelCase(header) }))

        .on("data", data => records.push(data))

        .on("end", async () => await saveToDB(getSysInfo(files[0]), records));

    }

  });

};

main();

我尝试.on("close")在结束后添加一个事件，但这也无济于事。

Answer 1

这是我解决它的方法。问题在于底层 MongoDB 连接本身。流没有错。从文件中读取记录后，将它们插入数据库，并在读取并插入所有文件中的所有记录后，关闭底层连接以结束程序。const getRecordsFromFile = fileName => {&nbsp; return new Promise((resolve, reject) => {&nbsp; &nbsp; const rows = [];&nbsp; &nbsp; fs.createReadStream(fileName)&nbsp; &nbsp; &nbsp; .pipe(csv({ skipLines: 7, mapHeaders: ({ header, index }) => _.camelCase(header) }))&nbsp; &nbsp; &nbsp; .on("data", row => rows.push(row))&nbsp; &nbsp; &nbsp; .on("end", () => resolve(rows));&nbsp; });};const main = async () => {&nbsp; const files = fs.readdirSync(dataPath);&nbsp; for (let i = 0; i < files.length; i++) {&nbsp; &nbsp; const records = await getRecordsFromFile(`${dataPath}/${files[i]}`);&nbsp; &nbsp; await loadRecordsToDB(getSysInfo(files[i]), records);&nbsp; }&nbsp; mongoose.connection.close();};main();

Answer 2

对于 CLI，您必须使用 db.close() 关闭 mongodb 连接/您可以关闭。句法：// any other clean ups&nbsp; &nbsp; mongoose.connection.close(function () {&nbsp; &nbsp; &nbsp; console.log('Mongoose connection disconnected');&nbsp; &nbsp; });重构代码：const util = require("util");const readDir = util.promisify(fs.readdir);const readCSV = () => {&nbsp; return new Promise((res, rej) => {&nbsp; &nbsp; let records = [];&nbsp; &nbsp; fs.createReadStream(filePath)&nbsp; &nbsp; &nbsp; .pipe(&nbsp; &nbsp; &nbsp; &nbsp; csv({&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; skipLines: 7,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; mapHeaders: ({ header, index }) => _.camelCase(header)&nbsp; &nbsp; &nbsp; &nbsp; })&nbsp; &nbsp; &nbsp; )&nbsp; &nbsp; &nbsp; .on("data", data => {&nbsp; &nbsp; &nbsp; &nbsp; records.push(data);&nbsp; &nbsp; &nbsp; })&nbsp; &nbsp; &nbsp; .on("error", error => {&nbsp; &nbsp; &nbsp; &nbsp; rej(data);&nbsp; &nbsp; &nbsp; })&nbsp; &nbsp; &nbsp; .on("end", () => {&nbsp; &nbsp; &nbsp; &nbsp; res(records);&nbsp; &nbsp; &nbsp; });&nbsp; });};const main = async () => {&nbsp; try {&nbsp; &nbsp; const files = await readDir(dataPath);&nbsp; &nbsp; await Promise.all(&nbsp; &nbsp; &nbsp; files.map(file => {&nbsp; &nbsp; &nbsp; &nbsp; return readCSV(`${dataPath}/${file}`) // read csv&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .then(csv => saveToDB(getSysInfo(file), csv)); // save in db&nbsp; &nbsp; &nbsp; })&nbsp; &nbsp; );&nbsp; } catch (error) {&nbsp; &nbsp; console.error(error);&nbsp; } finally {&nbsp; &nbsp; db.close(); // close mongo db&nbsp; }};main();

Answer 3

这是对您的答案的增强，它为 readStream 和两个await操作添加了错误处理，因此如果有任何错误，您的程序仍然可以以受控方式结束并正确关闭数据库：const getRecordsFromFile = fileName => {&nbsp; return new Promise((resolve, reject) => {&nbsp; &nbsp; const rows = [];&nbsp; &nbsp; fs.createReadStream(fileName)&nbsp; &nbsp; &nbsp; .pipe(csv({ skipLines: 7, mapHeaders: ({ header, index }) => _.camelCase(header) }))&nbsp; &nbsp; &nbsp; .on("data", row => rows.push(row))&nbsp; &nbsp; &nbsp; .on("end", () => resolve(rows));&nbsp; &nbsp; &nbsp; .on("error", reject);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // <==&nbsp; });};const main = async () => {&nbsp; const files = fs.readdirSync(dataPath);&nbsp; try {&nbsp; &nbsp; for (let i = 0; i < files.length; i++) {&nbsp; &nbsp; &nbsp; const records = await getRecordsFromFile(`${dataPath}/${files[i]}`);&nbsp; &nbsp; &nbsp; await loadRecordsToDB(getSysInfo(files[i]), records);&nbsp; &nbsp; }&nbsp; } catch(e) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;// <==&nbsp; &nbsp; &nbsp; console.log(e);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // <==&nbsp; } finally {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // <==&nbsp; &nbsp; &nbsp; // make sure we always close the connection&nbsp; &nbsp; &nbsp; mongoose.connection.close();&nbsp; }};main();