Node.js async.queue() 方法

async 模块是为在 NodeJS 中使用异步JavaScript 而设计的。 async.queue 返回一个能够并发处理的队列对象，即一次处理多个项目。

如何使用 async.queue？

• 第 1 步：创建一个 package.json 文件。运行以下命令会创建一个 package.json 文件。
  

npm init

• 第 2 步：安装异步模块。可以使用以下命令安装异步模块。
  

npm i async

• 第 3 步：导入异步模块。可以使用以下命令导入异步模块。
  

const async = require('async')

• 第 4 步：使用 async.queue 模块。 async.queue 的语法。
  

const queue = async.queue('function', 'concurrency value')

参数函数在添加到队列的元素上执行。并发值告诉队列，在特定时间要处理的元素数。

示例：请查看以下示例以更好地理解。

Javascript

// Defining The queue
const queue = async.queue((task, completed) => {
    // Here task is the current element being
    // processed and completed is the callback function
     
    console.log("Currently Busy Processing Task " + task);
     
    // Simulating a complex process.
    setTimeout(()=>{
        // Number of elements to be processed.
        const remaining = queue.length();
        completed(null, {task, remaining});
    }, 1000);
 
}, 1);
 
// The concurrency value is set to one,
// Which means that one element is being
// Processed at a particular time

async.queue 中的重要方法和属性：

push(element, callback) : push 方法用于将元素添加到队列的尾部。以下代码演示了 push 方法的工作原理。

Javascript

// Takes in two parameters, the item being pushed and
// the callback function
// Here the item is the element being added
// and other is the callback function with error
// and data property(destructured)
 
queue.push(item, (error, {item, remaining}) => {
  if(error){
      console.log(`An error occurred while processing task ${task}`);
  } else {
      console.log(`Finished processing task ${task}
             . ${remaining} tasks remaining`);
  }
});

1. 
   

length()： length 方法返回队列中当前存在的元素数。以下代码演示了长度方法的工作原理。

Javascript

console.log(queue.length());

1. 
   

started 属性： started 属性返回一个布尔值，表示队列是否已经开始处理数据。以下代码演示了 started 属性的工作原理。

Javascript

// Returns true if the queue has started processing the data else false
console.log(queue.started)

1. 
   

unshift(element, callback) ： unshift方法和push方法类似，但是元素被添加到队列的头部，表示要处理的元素是重要的。以下代码演示了 unshift 方法的工作原理：-

Javascript

// Takes in two parameters, the item being pushed
// and the callback function
// Here the item is the element being added
// and other is the callback function with error
// and data property(destructured)
 
queue.unshift(item, (error, {item, remaining}) => {
  if(error){
   console.log(`An error occurred while processing task ${task}`);
  } else {
   console.log(`Finished processing task ${task}. ${remaining} tasks remaining`);
  }
});

1. 
   

drain() 方法：当队列执行完所有任务时，drain 方法运行一个回调函数。下面的代码演示了 drain 方法的工作原理。

注意： drain 方法仅在所描述的函数是箭头函数时才有效。

Javascript

// Executes when the queue is done processing all the items
queue.drain(() => {
    console.log('Successfully processed all items');
})

pause() 方法： pause 方法暂停队列中元素的执行，直到调用 resume函数。以下代码演示了 pause 方法的工作原理。

Javascript

// Pauses the execution of the queue
queue.pause()

1. 
   

resume() 方法： resume 方法恢复队列中元素的执行。以下代码演示了 resume 方法的工作原理。

Javascript

// Resumes the execution of the queue
queue.resume()

1. 
   

kill() 方法： kill 方法从队列中移除所有元素，drain 方法的回调函数并强制其进入空闲状态。下面的代码演示了 kill 方法的工作原理。

Javascript

// Forces the queue into idle mode
// and removes the drain callback
queue.kill()

1. 
   

idle() 方法： idle() 方法返回一个布尔值，指示队列是空闲还是正在处理某事。下面的代码演示了 idle 方法的工作原理。

Javascript

// Returns whether the queue is idle or not
queue.idle()

1. 
   

完整代码：以下代码完整演示了 async.queue 的实际使用方式。

Javascript

// Importing the async module
const async = require('async');
 
// Creating a tasks array
const tasks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
 
// Defining the queue
const queue = async.queue((task, completed) => {
    console.log("Currently Busy Processing Task " + task);
     
    // Simulating a Complex task
    setTimeout(()=>{
        // The number of tasks to be processed
        const remaining = queue.length();
        completed(null, {task, remaining});
    }, 1000);
 
}, 1); // The concurrency value is 1
 
 
// The queue is idle as there are no elements
// for the queue to process
console.log(`Did the queue start ? ${queue.started}`)
 
// Adding the each task to the queue
tasks.forEach((task)=>{
 
    // Adding the 5th task to the head of the
    // queue as it is deemed important by us
 if(task == 5){
    queue.unshift(task, (error, {task, remaining})=>{
      if(error){
       console.log(`An error occurred while processing task ${task}`);
      }else {
       console.log(`Finished processing task ${task}. ${remaining} tasks remaining`);
      }
    })     
        // Adding the task to the tail in the order of their appearance
 } else {
      queue.push(task, (error, {task, remaining})=>{
       if(error){
        console.log(`An error occurred while processing task ${task}`);
       }else {
        console.log(`Finished processing task ${task}. ${remaining} tasks remaining`);
      }
      })
    }
});
 
 
// Executes the callback when the queue is done processing all the tasks
queue.drain(() => {
    console.log('Successfully processed all items');
})
 
// The queue is not idle it is processing the tasks asynchronously
console.log(`Did the queue start ? ${queue.started}`)

输出：