使用 unshift 和 pop 序列处理对象数组

Question

我有一个要在其中推送对象的数组，我正在使用array.unshift(object) 将对象添加到数组的开头。

我正在使用 var object = array.pop() 从数组中检索元素，它正在从数组末尾删除对象。

现在我的数组被快速填充，这个数组中的多个对象每秒都被推送。每次在数组中推送一个新元素时，我都需要弹出数组并处理它并进一步发送它，这需要时间。

那么每次间隔1秒后调用一个方法然后从数组中弹出一个元素并处理它是不是好方法，但是如果从数组中处理对象需要超过1秒，那么下一步对流程方法的调用也应该延迟，简而言之，应该在每个间隔之后调用流程数组方法，但如果之前的调用仍在处理中，则应该等待。

实现它的更好方法是什么？

示例代码：

angular.module('myApp')
.service('DataService', ['$rootScope', 'UpdateService', function($rootScope, UpdateService)
{
    // To store array of JSON updates    
    var arrayOfUpdates = [];

    // To store callbacks to which we need to send updates after processing 
    var arrayofCallbacks = [];

    // Method is getting called when there are updates from "UpdateService" server 
    this.onDataUpdated = function (jsonObj)
    {
        // This will add this object in the array
        arrayOfUpdates.unshift(jsonObj);
    }

    // This method will process data and send it further 
    function processData()
    {
         // Get the last element from the array 
         var jsonObj = arrayOfUpdates.pop();

         // Now process this JSON data  
         var data = doSomeCalculations(jsonObj); 

         // There can be 10-20 callbacks at a time   
         for(var index=0;index<arrayofCallbacks.length;index++)
         {
               var object = arrayofCallbacks[index]; 
               object.onUpdatedData(data);
         }       

         // After this pop next element and process and send to registered callbacks 
         // I can not call "processData()" again as it can create loop 
         // Also calling it after interval might call it when it has not completed processing 
    }
});

Answer 1

在我看来，您正在实施一个队列。您希望能够推到队列的后面（unshift），这可能会很快发生。您还想处理队列前面的项目 (pop)，但处理阶段需要时间。我的猜测是这些进程是异步运行的，因此您最终会同时发生许多处理步骤，并且您的数据会遇到一些类似竞争的情况。

除非您需要主动检查队列中的内容（在您的处理功能之外），否则我建议您将队列全部移除，并使用承诺链代替队列。

为此，每当触发传入数据事件时，您只需将数据传递到您的承诺链，直到所有先前的处理完成后才会执行。一些伪js：

// the promise chain (initialised to automatically trigger then())
var worker = Promise.resolve();

// handle incoming data
function incomingData(data){
    // process this data AFTER all other processing and update the chain
    worker = worker.then(last => processData(data));
}

// process data
function processData(data){
    // do work on the data and return a promise if work is async
    return ...
}

---

更新：我建议的第一件事是阅读Promises。但我会在下面做更深入的解释：

1. 设置工作器。通过将变量设置为Promise.resolve()，您实际上是在创建一个已经解决（完成）的承诺。把它想象成一个空队列，它已经完成了它需要做的所有工作。

2. 当您获得新的传入数据时更新承诺链。如果你在里面放一个函数，比如myPromise.then(myFunc)，myFunc 将在myPromise 被解析之前不会运行，这取决于程序员。它可能会在一些 api 调用返回后或在其他一些工作后得到解决。你可以认为对worker的更新如下：

   想象每次新数据到达时运行processData(data)，传入接收到的data。现在想象在前一个完成之前无法运行该功能。这就是上面代码中发生的事情。所以是的，传入的数据被存储了。

Answer 2

假设你的耗时任务是异步的，你可能会想出一个带有如下承诺的递归异步流控制；

我们每 250 毫秒将一个介于 1 和 10 之间的随机数移至 arr 数组，同时每 1000 毫秒弹出一个。定义的限制是 10 个项目，一旦我们达到 limit 值，取消移动停止。

function doStgAndPop(a){
  return new Promise((resolve,reject) => { var id = setTimeout(_ => a.length > 0 ? resolve(a.pop()) : reject(id),1000)});
}

function doStgAndUnshift(a){
  return new Promise((resolve,reject) => { var id = setTimeout(_ => a.length < limit ? (a.unshift(~~(Math.random()*10+1)),resolve(a))
                                                                                     : reject(id),250)});
}

function runnerPop(a){
  doStgAndPop(a).then(v => (console.log("pop out resulted:",v),runnerPop(a)))
                .catch(e => (console.log("array is now empty"),clearSetTimeout(e)));
}
function runnerUnshift(a){
  doStgAndUnshift(a).then(v => (console.log("new array after unshift",JSON.stringify(v)),runnerUnshift(a)))
                    .catch(e => (console.log("array limit:",limit,"reached"),clearSetTimeout(e)));
}

var limit = 10,
      arr = [];

runnerUnshift(arr);
runnerPop(arr);

我正在使用原生 ES6 承诺。 doStgAndPop 返回一个承诺并运行一个异步函数 (setTimeout)，如果 arr.length > 0 或拒绝操作，该函数会挂起 1000 毫秒并从 arr 弹出一个项目。doStgAndUnshift 非常相似，但每 250 毫秒取消一个项目直到达到限制，一旦达到限制，它就会拒绝。 runnerPop 和 runnerUnshift 是递归函数，它们分别调用 doStgAndPop 和 doStgAndUnshift 并收到一个承诺。在then 阶段，我们递归调用运行器函数，在catch 阶段，我们完成操作。