nodejs 等到循环中的所有 MongoDB 调用完成

Posted 2023-03-07

【中文标题】nodejs 等到循环中的所有 MongoDB 调用完成

【英文标题】：nodejs wait until all MongoDB calls in loop finish

【发布时间】：2016-02-13 08:32:48

【问题描述】：

我正在逐行读取 CSV 文件中的数据流，并在每一行上调用 findOne MongoDB，我如何才能等到每一行的所有 mongo 调用完成后再运行下一个函数？

我见过 Promises 可以做到吗？但是我发现 Promises 非常难以理解。而且我发现的所有例子似乎都没有涵盖我正在尝试的内容。 ：/

var validProducts = [];

fs.createReadStream(req.file.path)
  .pipe(csvStream)
  .on('error', function (err) 
    console.error(err);
  )
  // loop through all rows
  .on('data', function (data) 
    if (data.size === 'a3') 
      ProductModel.findOne( sku: data.sku , function (err, product) 
        if (product !== null) 
          product.size = data.size;
          product.save();
          validProducts.push(product);
        
      );
    
  );

// on finish make call to other function
socket.emit( 'status': 'complete' );
otherFunction(validProducts);

on('finish') 或 on('end') 只会在数据流的末尾调用，而不是在 Monogo 调用之后。

如果我可以使用承诺，有人可以解释一下吗？

【参考方案1】：

您可以使用Q API 女巫让您做出承诺。有一个有趣的函数可以让你等待一系列 Promise 被解决。以下是如何使用Q.all 解决问题的示例：

var validProducts = [];
var promises = [];

function handleData(data) 
    if (data.size === 'a3') 

        var deferred = Q.defer();

        ProductModel.findOne( sku: data.sku , function (err, product) 
            if (err) 
                deferred.reject(new Error(err));
            

            if (product) 
                product.size = data.size;
                product.save();
                deferred.resolve(product);
                validProducts.push(product);
            

        );

        promises.push(deferred.promise);

    


function handleEnd() 
    Q.all(promises).done(function (values) 
        socket.emit( 'status': 'complete' );
        otherFunction(validProducts);
    );


fs.createReadStream(req.file.path)
  .on('data', handleData)
  .on('end', handleEnd);

【讨论】：

:P 我一个人就这么接近这个了，。但没有正确设置延迟的东西。

这根本不会wait till all the mongo calls from each row are complete before I run the next function

也许我的逻辑有缺陷？如果我错了，请纠正我，但是当没有更多数据可以从文件中读取时，我们正在等待所有的承诺在执行指定的函数之前得到解决。

如果我的想法是正确的，则在将任何内容添加到 promises 数组之前，流将 end。所以Q.all(promises) 会立即执行。 ://

其实一开始有错字！我编辑了它，我认为它会被忽视。很高兴 ;)

【参考方案2】：

使用pause/resume

  .on('data', function (data) 
    if (data.size === 'a3') 
      this.pause(); // pause it
      var stream = this; // save 'this'
      ProductModel.findOne( sku: data.sku , function (err, product) 
        if (product !== null) 
          product.size = data.size;
          product.save();
          validProducts.push(product);
        
        stream.resume(); //resume it
      );
    
  );

【讨论】：

这不是使函数同步，而不是异步吗？

关闭。但是.resume() 和.push() 需要在来自.save() 的回调中发生。也不确定this 是否暴露，因此您可能需要先从外部抓取流。

这是一个数据库的 I/O 函数。节点中几乎所有的 I/O 都是异步的，或者应该是异步的。

我实际上认为我不知道问题是什么：D - 我认为 each row 正在由 on('data' 函数处理

I/O 是异步的，但是如果流暂停直到前一个完成，则一次只会发生一个 mongo 调用，这使得它同步（技术上），同时也（技术上）回答我的问题..ish。