promise.all 在 forEach 循环中——所有的东西同时触发

Question

在 Node 应用程序中，我需要以同步方式遍历某些项目，但循环内的某些操作是异步的。我的代码现在看起来像这样：

someAPIpromise().then((items) => {
   items.forEach((item) => {
      Promise.all[myPromiseA(item), myPromiseB(item)]).then(() => {
         doSomethingSynchronouslyThatTakesAWhile();
      });
    }
}

当 items 是 1 的数组时，这会产生奇迹。但是，一旦有多个项目，promise.all() 将立即为数组中的每个项目触发，而无需等待循环中的操作结束。

综上所述...我如何确保数组中每个项目的整个操作是运行同步的（即使某些操作是异步的并且 return 是一个承诺）？

非常感谢！

N

Answer 1

您正在构建多个承诺，但它们都是异步的。您构建了 Promise1、Promise2、Promise3……但是一旦它们在野外，它们就会同时开火。如果你想要同步行为，你必须将它们链接在一起，以便 Promise1 的 .then() 执行 Promise2 等等。过去我为此使用 Array.reduce。

someAPIpromise().then((items) => {
    items.reduce((accumulator, current) =>
        accumulator.then(() =>
             Promise.all[myPromiseA(item), myPromiseB(item)]).then(() => 
                 doSomethingSynchronouslyThatTakesAWhile();
             )
        )
    , Promise.resolve());

如果你愿意，你可以把它写成一个辅助函数，这可能会让事情变得更清楚。

function execSequentially (arr, func) {
    return arr.reduce(
        (accumulator, current) => accumulator.then(() => func(current)), 
        Promise.resolve());
}

该函数执行为

execSequentially(items, item => console.log(item));

当然可以将 console.log 替换为您想要执行的操作。

辅助函数方法的更改侵入性也较小。应用于您的原始代码的助手：

someAPIpromise().then((items) => {
   execSequentially(items, (item) =>
      Promise.all[myPromiseA(item), myPromiseB(item)]).then(() => {
         doSomethingSynchronouslyThatTakesAWhile();
      });
   );
});

Answer 2

您应该可以删除 .forEach()；使用 Array.prototype.reduce() 到 return 的数组 Promise 值到 Promise.all()。如果带有 items 的元素是一个函数，则调用函数，否则在 Promise.resolve() 中换行，这应该 return 结果与 items 数组

中的顺序相同

见Promise.all()

Promise.all passes an array of values from all the promises in the iterable object that it was passed. The array of values maintains the order of the original iterable object, not the order that the promises were resolved in. If something passed in the iterable array is not a promise, it's converted to one by Promise.resolve.

var arr = [1, // not asynchronous
  function j() {
    return new Promise(function(resolve) {
      setTimeout(function() {
        resolve(2)
      }, Math.floor(Math.random() * 10000))
    })
  }, // asynchronous
  3, // not asynchronous
  function j() {
    return new Promise(function(resolve) {
      setTimeout(function() {
        resolve(4)
      }, Math.floor(Math.random() * 3500))
    })
  }, // asynchronous
  5, // not asynchronous
  Promise.resolve(6), // asynchronous
  7
];

Promise.all(arr.reduce(function(p, next) {
    var curr = Promise.resolve(typeof next === "function" ? next() : next);
    return p.concat.apply(p, [curr.then(function(data) {
      console.log(data);
      return data
    })]);
  }, []))
  .then(function(data) {
    console.log("complete", data)
  })

另一种方法是使用 Array.prototype.shift()、Promise.resolve()、.then()、递归

function re(items, res) {
  if (items.length) {
    var curr = items.shift();
    return Promise.resolve(
      typeof curr === "function" 
      ? curr() 
      : curr
    ).then(function(data) {
      // values from `arr` elements should be logged in sequential order
      console.log(data);
      res.push(data)
    }).then(re.bind(null, items, res))
  } else {
    return ["complete", res]
  }
}

var _items = arr.slice(0);

re(_items, [])
.then(function(complete) {
  console.log(complete)
})

var arr = [1, // not asynchronous
  function j() {
    return new Promise(function(resolve) {
      setTimeout(function() {
        resolve(2)
      }, Math.floor(Math.random() * 10000))
    })
  }, // asynchronous
  3, // not asynchronous
  function j() {
    return new Promise(function(resolve) {
      setTimeout(function() {
        resolve(4)
      }, Math.floor(Math.random() * 3500))
    })
  }, // asynchronous
  5, // not asynchronous
  Promise.resolve(6), // asynchronous
  7
];

function re(items, res) {
  if (items.length) {
    var curr = items.shift();
    return Promise.resolve(
      typeof curr === "function" 
      ? curr() 
      : curr
    ).then(function(data) {
      // values from `arr` elements should be logged in sequential order
      console.log(data);
      res.push(data)
    }).then(re.bind(null, items, res))
  } else {
    return ["complete", res]
  }
}
var _items = arr.slice(0);
re(_items, [])
  .then(function(complete) {
    console.log(complete)
  })

Answer 3

好吧......我们能够让它工作的方式：array.reduce() 在 Promises 的帮助下。最终结果：

myAsyncAPIcall.then(items => {
    items.reduce((current, nextItem) => {
        return current.then(() => {
          return new Promise(res => {
             Promise.all([myPromiseA(nextItem), myPromiseB(nextItem]).then(() => {
               someSynchronousCallThatTakesAWhile(nextItem);
               res();
             }).catch(err => {
                   console.log(err);
             });
          });
        });
    }, Promise.resolve())
})

它的工作方式是，通过将数组的每一项包装在自己的 Promise(resolve, reject) 中，我们可以确保每次迭代都是同步的运行，因为一次迭代的完成将触发需要解决下一个 Promise，依此类推。在每个 promise 解析中，可以根据需要异步启动调用，只要知道在完成之前它们的范围仅限于父 promise。

希望对大家有所帮助！

Answer 4

如何保留 forEach...

var stopAllProcessingOnServerLowValue= false;

function someAPIpromise(){
    var arr = [
        {id:123, urlVal:null},
        {id:456, urlVal:null},
        {id:789, urlVal:null},
        {id:101112, urlVal:null}
    ];

    return new Promise(function(resolve){
        setTimeout(function(){
            resolve(arr)
        }, 3000);
    })
}

function extractSomeValueRemotely(url){
    return new Promise(function(resolve, reject){
        console.log("simulate an async connection @ %s to request a value", url);
        setTimeout(function(){
            var someRandom = Math.round(Math.random()*7) + 1;
            console.log("%s responded with %s", url, someRandom);
            if(someRandom > 4){
                resolve(someRandom);
            }
            else{
                var issue = "Urls result is too low ("+someRandom+" <= 4).";
                console.warn(issue+".It will be set to -1");
                if(stopAllProcessingOnServerLowValue){
                    reject(issue+".Operation rejected because one or mole server results are too low ["+someRandom+"].");
                }
                else{
                    resolve(-1);
                }
            }
        }, 1500*Math.round(Math.random()*7) + 1);
    });
}

function addAnotherExtraParamToItem(_item){
    return new Promise(function(resolve, reject){
        setTimeout(function(){
            console.log("setting extra2 on %s", _item.id);
            _item['extra'] = "additional_processing_"+_item.id;
            resolve(_item);
        }, 1500*Math.round(Math.random()*5) + 1);
    });
}

function addOrderIndexToItem(_item, _order){
    return new Promise(function(resolve, reject){
        setTimeout(function(){
            console.log(">> setting order %s on %s",_order,  _item.id);
            _item['order'] = _order;
            resolve(_item);
        }, 1500*Math.round(Math.random()*3) + 1);
    });
}

someAPIpromise().then(function(items){

    var perItemPromises = [];
    items.forEach(function(item, idx){

        perItemPromises.push(

            new Promise(function(pulseItemResolve, pulseItemReject){
                var itemStepsPromises =  [];
                itemStepsPromises.push(addAnotherExtraParamToItem(item));

                itemStepsPromises.push(extractSomeValueRemotely("http://someservice:777/serve-me")
                    .catch(
                        function(reason){
                            //the entire item will be rejected id
                            pulseItemReject(reason);
                        })
                );

                itemStepsPromises.push(addOrderIndexToItem(item, idx));

                //promise that ensure order of execution on all previous async methods
                Promise.all(itemStepsPromises).then(function(values){
                    //0 - first is result from addAnotherExtraParamToItem
                    var theItem = values[0]; //it returns the item itself
                    //urlVal has not been set yet

                    // 1 - second promise return the url result
                    var serverResult = values[1];

                    //2 - third promise add the order index but we do not care to inspect it because theItem reference in value[0] has been already updated.
                    // console.info(values[2]);

                    //sets the url result in the item
                    theItem.urlVal = serverResult;
                    console.log("urlVal set to:", theItem.urlVal);

                    //resolve the prepared item
                    pulseItemResolve(theItem);

                });
            })
                .catch(function(reason){
                    //escalate error
                    throw new Error(reason);
                })
        )

    });

    Promise.all(perItemPromises).then(function(resultsInAllItems){
        console.info("Final results:");
        console.info(resultsInAllItems);
    }).catch(function(finalReject){
        console.error("Critical error:",finalReject);
    })


});

Answer 5

经过大量研究，这里是我的最终答案...

我已经阅读了一堆解决方案，因为它们具有有用的纯 JavaScript（无插件）-Promise Iterator- 可以轻松使用（一行）所有在我的项目中，最后我找到了 Salketer 的 this solution：

function one_by_one(objects_array, iterator, callback) {
    var start_promise = objects_array.reduce(function (prom, object) {
        return prom.then(function () {
            return iterator(object);
        });
    }, Promise.resolve()); // initial
    if(callback){
        start_promise.then(callback);
    }else{
        return start_promise;
    }
}

有关详细信息和用法示例，请访问 the link。

它还允许直接处理回调。

在与 Promise 迭代和测试来自许多问题、博客和官方网站的多个解决方案进行了多日的斗争之后，这简直是我发现的最合乎逻辑和可重用的方法。

如果您也在为确定的答案而苦苦挣扎，请试一试。

promise.all 在 forEach 循环中——所有的东西同时触发

promise.all inside a forEach loop — everything firing at once

javascript

asynchronous

node.js

promise

es6-promise