如何使用node.js http服務器從mongodb返回大量行？

Question

我有一個用戶數據庫mongodb，我想通過JSON的REST接口導出。問題是，在最壞的情況下，返回的行數已經超過200萬。

首先我想這

var mongo = require('mongodb'), 
    Server = mongo.Server, 
    Db = mongo.Db; 
var server = new Server('localhost', 27017, {auto_reconnect: true}); 
var db = new Db('tracking', server); 
var http = require('http'); 

http.createServer(function (request, response) { 
    db.collection('users', function(err, collection) { 
    collection.find({}, function(err, cursor){ 
     cursor.toArray(function(err, items) { 
     output = '{"users" : ' + JSON.stringify(items) + '}'; 

     response.setHeader("Content-Type", "application/json"); 
     response.end(output); 
     }); 
    }); 
    }); 
}).listen(8008); 
console.log('Server running at localhost:8008'); 

運行內存時失敗。該示例使用node-mongodb-native驅動程序和基本的http軟件包。

FATAL ERROR: CALL_AND_RETRY_2 Allocation failed - process out of memory

（注意，我用的需要，限制的結果參數真實的情景，但這個例子查詢所有這些這是最壞的情況下，無論）

數據本身很簡單，就像

{ "_id" : ObjectId("4f993d1c5656d3320851aadb"), "userid" : "80ec39f7-37e2-4b13-b442-6bea57472537", "user-agent" : "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 1.1.4322)", "ip" : "127.0.0.1", "lastupdate" : 1335442716 }

我也嘗試過類似

while(cursor != null) 
{ 
    cursor.nextObject(function(err, item) { 
    response.write(JSON.stringify(item)); 
    }); 
} 

但跑出來的記憶也。

我該如何繼續？應該有一種方法可以逐行地對數據進行流式處理，但是我一直無法找到合適的示例。由於外部應用程序需求，分頁數據不存在問題。我想將數據寫入一個文件然後發佈，但這會導致不需要的io。

Answer 1

天然MongoDB驅動器的方法cursor.streamRecords()已過時， 方法stream()更快。

我已經解析acatalog的4000萬行的文件沒有問題與Mongodb + stream() + process.nextTick()

Answer 2

那麼，我不再使用mongodb原生JavaScript驅動程序，但在mongoose有很好的流實現。

這兩個驅動程序的語法非常相似。你可以用貓鼬做到這一點：

response.setHeader("Content-Type", "application/json"); 
var stream = collection.find().stream(); 
stream.on('data', function(doc) { 
    response.write(doc); 
}); 
stream.on('close', function() { 
    response.end(); 
}); 

Answer 3

類似的東西應該工作。如果不是，你應該在mongodb-native bug tracker中打開一個問題。

http.createServer(function (request, response) { 
    db.collection('users', function(err, collection) { 
    collection.find({}, function(err, cursor){ 
     response.setHeader("Content-Type", "application/json"); 
     cursor.each(function(err, item) { 
     if (item) { 
      response.write(JSON.stringify(item)); 
     } else { 
      response.end(); 
     } 
     }); 
    }); 
    }); 
}).listen(8008); 

PS：這只是一個存根，我的意思是我不記得確切的語法，但它是你要找的each功能。

Answer 4

我發現node-mongodb-native Cursor對象也有一個流選項（與collection.find().streamRecords()一起使用）作爲記錄，即使沒有在github page of the driver中提及。請參閱Cursor source code並搜索「streamRecords」。

在代碼弄成這樣結束：

db.collection('users', function(err, collection) { 
    var first = true; 

    response.setHeader("Content-Type", "application/json"); 
    response.write('{"users" : ['); 

    var stream = collection.find().streamRecords(); 

    stream.on('data', function(item) { 
    var prefix = first ? '' : ', '; 
    response.write(prefix + JSON.stringify(item)); 
    first = false; 
    }); 
    stream.on('end', function() { 
    response.write(']}'); 
    response.end(); 
    }); 
}); 

Answer 5

一個小模塊做，使用節點的stream.Transform類：

var stream = require('stream'); 

function createCursorStream(){ 

    var cursorStream = new stream.Transform({objectMode:true}); 

    cursorStream._transform = function(chunk,encoding,done){ 
     if(cursorStream.started){ 
      cursorStream.push(', ' + JSON.stringify(chunk)); 
     }else{ 
      cursorStream.push('[' + JSON.stringify(chunk)); 
      cursorStream.started = true; 
     } 
     done(); 
    }; 

    cursorStream._flush = function(done){ 
     cursorStream.push(']'); 
     done(); 
    }; 

    return cursorStream; 
} 

module.exports.streamCursorToResponse = function(cursor,response){ 
    cursor.stream().pipe(createCursorStream()).pipe(response); 
}; 

可以改變JSON.Stringify部分做任何其他類型的「即時」轉換來自mongodb光標的對象，並保存一些內存。