Node.js流写入MongoDB - 关注性能

Question

我需要读取数千行的日志文件并将每行写入Mongo数据库。我正在使用节点流读取文件。我正在使用'split'npm软件包将文件分成'行'。由于网络的考虑，MongoDB的写入将比读取的日志文件花费更多的时间。

我的核心代码如下所示：

var readableStream = fs.createReadStream(filename); 

      readableStream 
       .pipe(split()) // This splits the data into 'lines' 
       .on('data', function (chunk) { 

        chunkCount++; 
        slowAsyncFunctionToWriteLogEntryToDatabase(chunk); // This will take ages 

       }) 
       .on('end', function() { 
        // resolve the promise which bounds this process 
        defer.resolve({v:3,chunkCount: chunkCount}) 

       }); 

我需要担心的MongoDB系统将写入次数敲定排队？据推测，节点管道背压机制不会知道大量数据库写入正在排队？有什么方法可以“减慢”可读流，以便它在读取日志文件中的下一行之前等待每个MongoDB插入完成？我是否不必要地担心？

Answer 1

由于使用pause()和resume()似乎有一些问题。我将编写另一个选项，使用Transform流。

var Transform = require('stream').Transform; 

var myTransform = new Transform({ 
    transform(chunk, encoding, cb) { 
     chunkCount++; 

     syncFunctionToWriteLogEntryWithCallback(chunk, function() { 
     cb(); 
     }); 
    }, 

    flush(cb) { 
     chunkCount++; 
     syncFunctionToWriteLogEntryWithCallback(chunk, function() { 
     cb(); 
     }); 
    } 
}); 

readableStream 
     .pipe(split()) 
     .pipe(myTransform); 

使用转换流允许您在完成流处理后提供回调。

Answer 2

您可以在可读流中使用pause method在您将块写入到mongodb时停止流。

readableStream 
      .pipe(split()) // This splits the data into 'lines' 
      .on('data', function (chunk) { 

       readableStream.pause() 

       chunkCount++; 

       syncFunctionToWriteLogEntryWithCallback(chunk, function() { 
        readableStream.resume(); 
       }); 

      }) 
      .on('end', function() { 
       // resolve the promise which bounds this process 
       defer.resolve({v:3,chunkCount: chunkCount}) 

      }); 

我不认为在这种情况下MongoDB会出现严重问题。