Блог

Я пытаюсь написать немного JS, который будет считывать файл и записывать его в поток. Дело в том, что файл очень большой, и поэтому мне приходится читать его по крупицам. Кажется, что у меня не должно быть нехватки памяти, но я это делаю. Вот код:

 var size = fs.statSync("tmpfile.tmp").size;

var fp = fs.openSync("tmpfile.tmp", "r");

for(var pos = 0; pos < size; pos  = 50000){
    var buf = new Buffer(50000),
        len = fs.readSync(fp, buf, 0, 50000, (function(){
            console.log(pos);
            return pos;
        })());

    data_output.write(buf.toString("utf8", 0, len));

    delete buf;
}

data_output.end();
  

По какой-то причине он достигает 264900000, а затем выбрасывает FATAL ERROR: CALL_AND_RETRY_2 Allocation failed - process out of memory . Я бы предположил, что data_output.write() вызов заставит его записать данные data_output , а затем удалить их из памяти, но я могу ошибаться. Что-то заставляет данные оставаться в памяти, и я понятия не имею, что это будет. Любая помощь была бы весьма признательна.

У меня была очень похожая проблема. Я читал в очень большом файле csv с 10 строками и записывал его эквивалент в формате json. Я увидел в диспетчере задач Windows, что мой процесс использует> 2 ГБ памяти. В конце концов я понял, что выходной поток, вероятно, был медленнее, чем входной поток, и что выходной поток буферизовал огромное количество данных. Я смог исправить это, приостановив instream каждые 100 записей в outstream и дождавшись, пока outstream опустеет. Это дает время для того, чтобы outstream догнал instream. Я не думаю, что это имеет значение для этого обсуждения, но я использовал ‘readline’ для обработки файла csv по одной строке за раз.

Я также выяснил по пути, что если вместо записи каждой строки в outstream я объединю 100 или около того строк вместе, а затем запишу их вместе, это также улучшит ситуацию с памятью и ускорит работу.

В конце концов, я обнаружил, что могу выполнить передачу файлов (csv -> json), используя всего 70 МБ памяти.

Вот фрагмент кода для моей функции записи:

 var write_counter = 0;
var out_string = "";
function myWrite(inStream, outStream, string, finalWrite) {
    out_string  = string;
    write_counter  ;
    if ((write_counter === 100) || (finalWrite)) {
        // pause the instream until the outstream clears
        inStream.pause();
        outStream.write(out_string, function () {
            inStream.resume();
        });
        write_counter = 0;
        out_string = "";
    }
}
  

Вы должны использовать трубы, такие как:

 var fp = fs.createReadStream("tmpfile.tmp");
fp.pipe(data_output);
  

Для получения дополнительной информации ознакомьтесь с: http://nodejs.org/docs/v0.5.10/api/streams.html#stream.pipe

РЕДАКТИРОВАТЬ: проблема в вашей реализации, кстати, заключается в том, что, делая это такими кусками, буфер записи не будет очищен, и вы будете читать весь файл, прежде чем записывать большую его часть обратно.

Согласно документации, data_output.write(...) вернется true , если строка была сброшена, и false если этого не произошло (из-за переполнения буфера ядра). Что это за поток?

Кроме того, я (вполне) уверен, что это не проблема, но: как получилось, что вы выделяете новое Buffer на каждой итерации цикла? Разве не имело бы больше смысла инициализировать buf перед циклом?

Я не знаю, как реализованы синхронные файловые функции, но рассматривали ли вы возможность использования асинхронных? Это с большей вероятностью позволит выполнить сборку мусора и очистку ввода-вывода. Таким образом, вместо цикла for вы должны инициировать следующее чтение в функции обратного вызова предыдущего чтения.

Что-то в этом роде (обратите внимание также, что, согласно другим комментариям, я повторно использую буфер):

 var buf = new Buffer(50000),
var pos = 0, bytesRead;  

function readNextChunk () {
    fs.read(fp, buf, 0, 50000, pos,
      function(err, bytesRead){
        if (err) {
          // handle error            
        }
        else {
          data_output.write(buf.toString("utf8", 0, bytesRead));
          pos  = bytesRead;
          if (pos<size)
            readNextChunk();
        }
      });
}
readNextChunk();