仅在回调nodejs中完成一次调用异步函数

Question

我有以某种方式接收 100000 个数据集的代码。 然后有一个存储必须以一种只有在最后一次添加完成后才能开始下一个才能开始的方式访问。

以一种看起来像这样的同步方式..所以 add 方法会阻塞。

var data = [...]; //100000 datasets 
var syncstorage = require( 'syncstorage' ); // syncronous storage.

for( var i = 0 ; i < data.length() ; i++ ) {
    syncstorage.add( data[i] ); // will only return once stored
}

异步存储不会阻塞，但它会在回调完成时告诉您...

/////// async storage 

asyncstorage.add( data[i] , function(err) { /* can only execute another add once i get this response */ } )

我现在才想到这个：

var i = 0;
function execute() {
    if( i >= data.length()){
        return;
    }
    asyncstorage.add( data[i] , function(err) { i++; execute(); } )     
}

但是它会导致一个极端的调用堆栈

也许我需要一个事件发射器并在该回调中发出它？ 有点像资源包装器？ 这是如何解决的？ ...遗憾的是，我没有发现与此特定问题有关的堆栈溢出结果；

排除解决方案

• async.each
  • http://caolan.github.io/async/docs.html#.each
  • “此函数将 iteratee 并行应用于每个项目”
• 异步系列
  • http://caolan.github.io/async/docs.html#.series
  • 1. 它需要一个不是数据的函数数组
  • 1. 这将需要从回调中调用回调才能不准并行运行
  • 所以它可以使用，但只能用于不成比例的内存需求（函数）
  • 也不确定异步如何在内部处理这个问题（调用堆栈）

测试示例

var store = {add:function(d,cb){cb(null)}};
var d=[]; for(var i = 0 ; i < 100000; i ++) { d.push(i)}; d;
var async = require('async');
async.eachSeries(d,store.add);

不起作用！ 这是因为 async 假定 iteratee 函数中将存在一个事件发射器。 因此，像上面这样的简单测试类在 Maximum call stack size exceeded

中运行

Answer 1

使用承诺或Async

var async = require('async');

// Assuming that asyncstorage.add = function(chunk, callback) { ... }
async.eachSeries(data, asyncstorage.add, function(err) { 
    if (err)
        console.log(err);
    ...
});

---

删除Maximum call stack size exceeded 的决定是调用nextTick。它是“让 node.js 有机会清除堆栈 (read more)”。

改进的代码
异步

var async = require('async');

var store = {
    add: function(chunk, cb){
        res.push(chunk);
        cb(null);
    }
};
var data = []; 
for (var i = 0 ; i < 100000; i ++)  
    data.push(i);

var res = []; // for test result

async.eachSeries(data, 
    // on each iteration.
    function f(chunk, cb) {
        async.nextTick(function() {
            store.add(chunk, cb)
        });
    }, 
    // on done
    function(err) {
        console.log((err) ? err : ('done ' + res.length));
    }
);

事件发射器

var data = []; 
for (var i = 0; i < 100500; i++) 
    data.push(i);

var store = {
    add: function (chunk, cb) { cb(null); }
};

var EventEmitter = require('events').EventEmitter;
var e = new EventEmitter;

e.on('next', function(i) {
    if (i > data.length) 
        return console.log(i, 'done'); 

    setImmediate(function() { // clear stack
        store.add(data[i], () => e.emit('next', i + 1))
    });
})

e.emit('next', 0);