带有 csv-write-stream 的“错误：结束后写入”

Question

我确信这从根本上是对流如何工作的误解，但我是在碰壁。

我有一些 json 格式的传感器数据，我想使用 csv-write-stream 包附加到 csv 文件中。数据作为发布请求发送到节点服务器，目的是将其附加到 csv 文件中。它第一次很好地将一行写入 csv 文件，但是如果我尝试发送另一个发布请求，则会收到“错误：结束后写入”错误。

function write_csv(obj) {
    writer.pipe(fs.createWriteStream('tides.csv', { flags: 'a' }));
    writer.write(obj);
    writer.end();
};

如果我注释掉“writer.end()”它可以正常工作，但这最终不会引发内存错误吗？如果是这样，追加到 csv 文件并避免此错误的正确方法是什么？

编辑：这是整个 server.js 文件

const express = require('express');
const bodyParser = require('body-parser');
const path = require('path');
const exphbs = require('express-handlebars');
const fs = require('fs');
const csvWriter = require('csv-write-stream');

const writer = csvWriter({ sendHeaders: false });
const app = express();

app.set('views', path.join(__dirname, 'views'));
app.engine('handlebars', exphbs({ defaultLayout: 'main' }));
app.set('view engine', 'handlebars');

app.set('port', (process.env.PORT || 3000));

app.use(express.static(path.join(__dirname, 'public')));

app.use(bodyParser.urlencoded({ extended: false }))
app.use(bodyParser.json())

app.get('/', function (req, res) {
    res.render('home')
})

app.post('/test', function (req, res, next) {
    // console.log("post received");
    distance = req.body.distance;
    let result = test(distance);
    let result_str = JSON.stringify(result);
    res.end(result_str)
});

function write_csv(obj) {
    writer.pipe(fs.createWriteStream('out.csv', { flags: 'a' }));
    writer.write(obj);
    writer.end();
};

function test(dist) {
    let d = new Date();
    let YYYY = d.getFullYear();
    let MM = d.getMonth();
    let DD = d.getDate();
    let HH = d.getHours();
    let mm = d.getMinutes();
    let ss = d.getSeconds();
    let date = YYYY + ':' + (MM + 1) + ':' + DD + ':' + HH + ':' + mm + ':' + ss;
    let time_distance = { 'time': date, 'distance': distance };
    console.log(time_distance);
    write_csv(time_distance);
    return time_distance;
};


app.listen(app.get('port'), function () {
    console.log('Sever started on port ' + app.get('port'));
})

Answer 1

没有看到完整的代码，我可以想象你多次调用write_csv，因为你试图将多个对象写入该文件。

问题是您第一次调用 write_csv 时会结束 writer，这就是为什么您第二次调用它时会得到：

Error [ERR_STREAM_WRITE_AFTER_END]: write after end

---

function write_csv(obj) {
    writer.pipe(fs.createWriteStream('out.csv', { flags: 'a' }))
    writer.write()
    writer.end();
}
write_csv({ hello: 'world', foo: 'bar', baz: 'taco'});
// When you call it again, writer.end(); is already closed
// The following line will trigger the error
write_csv({ hello: 'world', foo: 'bar', baz: 'taco'});

你应该做的是，只有在你写完之后才关闭它。

const writer = csvWriter(); // Create new writer
// open file
writer.pipe(fs.createWriteStream('out.csv', { flags: 'a' }));

for(const obj of objects) // Write as many times as you wish
   writer.write(obj);

writer.end(); // I'm done writing.

---

现在您遇到的问题是，如果您尝试执行多个.writes，您将达到内存限制，因为您没有正确处理backpressure。

我建议阅读以下问题：

why does attempting to write large a large file cause js heap to run out of memory

要解决这个问题，您需要等待drain 事件被发出。

这是一个围绕csvWriter 的包装器，用于处理背压。

const fs = require('fs');
const csvWriter = require('csv-write-stream');

class Writer {

    constructor(file) {
        this.writer = csvWriter();
        this.writer.pipe(fs.createWriteStream(file, { flags: 'a' }));
    }

    write(obj) {
        // if .write returns false we have to wait until `drain` is emitted
        if(!this.writer.write(obj))
            return new Promise(resolve => this.writer.once('drain', resolve))

        return true;
    }

    end() {
        // Wrap it in a promise if you wish to wait for the callback.
        this.writer.end(); 
    }

}

(async() => {
    const writer = new Writer('out.csv');

    for(let i = 0; i < 1e8; i++) {
        const res = writer.write({ hello: 'world', foo: 'bar', baz: 'taco' });
        if(res instanceof Promise) {
            // You can remove this if, and leave just: await writer.write...
            // but the code will be slower
            await res; // This will wait for the stream to emit the drain event
        }
    }

    writer.end();

})();

---

更新：现在有了实际代码，上面的答案仍然有效，但是因为您在收到请求时正在写入文件。您可以选择是否打开文件一次，并在每个请求上写入，在服务器关闭时（或您选择时）关闭它。或者只是打开文件，写入文件，然后在每次请求时关闭它，

对于前者，您应该使用上面的答案，对于后者，您需要做的就是每次调用 write_csv 时创建一个新作者，而不是拥有一个全局作者。

function write_csv(obj) {
    // Create a new writer every time
    const writer = csvWriter({ sendHeaders: false }); 
    writer.pipe(fs.createWriteStream('out.csv', { flags: 'a' }));
    writer.write(obj);
    writer.end();
};

Answer 2

您可以尝试使用此代码，我认为它可以解决您的问题。在你的路由中使用这个函数并传递 json 数据。

首先安装这个包：npm i --save json2csv

const Json2csvParser = require("json2csv").Parser;
const fs = require("fs");

function csvConverter(jsonData, cb) {

  const json2csvParser = new Json2csvParser();
  const csv_data = json2csvParser.parse(jsonData);
  const file_name = "report";

  // TODO: Change file path accordingly
  const file_path = `/Users/public/csv_files/${file_name}.csv`;

  fs.appendFile(file_path, csv_data, err => {
    if (err) {
      console.log(err);
      cb(err, null);
      return;
    }
    const response = {
    msg: "successful",
    file_address: file_path
  };
    cb(null, response);
    return;
  });
}