Node.jsのasync.queue()メソッドとは？使い方と主要メソッドを徹底解説

asyncモジュールは、Node.jsアプリケーションで非同期処理を扱うためのさまざまな機能を提供するライブラリです。その中でもasync.queue()メソッドは、キュー（待ち行列）オブジェクトを返し、複数のタスクを同時並行で効率的に処理するために使用されます。

async.queue()のインストールと使い方

ステップ1：npmプロジェクトの初期化

まず、以下のコマンドを実行してnpmパッケージマネージャーを初期化します。

npm init

ステップ2：asyncモジュールのインストール

次に、以下のコマンドでasyncモジュールをインストールします。

npm install --save async

ステップ3：asyncモジュールの読み込み

プログラム内で以下の文を使ってasyncモジュールを読み込みます。

const async = require('async')

構文

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

パラメータ

• function … キューに追加された各要素に対して実行される関数を定義します。
• concurrency value（同時実行数） … 一度に処理する要素の数を指定します。

キューが持つ主なメソッドとプロパティ

async.queue()で生成されたキューには、非同期リクエストの処理中に利用できる便利なメソッドやプロパティが多数用意されています。

push(element, callback)

通常のキューと同様に、要素をキューの末尾に追加します。

queue.push(item, callback);

length()

現在キューに存在する要素数を返します。残りのタスク数を確認したいときに便利です。

queue.length()

started プロパティ

キューが処理を開始しているかどうかを真偽値（boolean）で返すプロパティです。

queue.started

unshift(element, callback)

push()と同じく要素を追加しますが、こちらはキューの先頭に挿入されます。優先的に処理したいタスクがある場合に使用します。

queue.unshift(item, callback)

drain コールバック

キュー内のすべてのタスクが完了したタイミングで呼び出されるコールバックです。すべての処理が終わったことを検知できます。

queue.drain(() => {
  console.log("すべてのタスクの実行が完了しました");
});

pause()

キュー内の残りの要素の処理を一時停止します。resume()が呼ばれるまで処理は再開されません。

queue.pause()

resume()

pause()によって停止されていた処理を再開します。

queue.resume()

kill()

キューに残っているすべての要素を削除し、キューをアイドル状態に強制します。

queue.kill()

idle()

キューがアイドル状態か、何かを処理中かどうかを真偽値で返します。

queue.idle

実装例

ここまでの内容を理解するために、実際のコード例を見てみましょう。

// asyncモジュールを読み込む
const async = require('async');

// 実行する全タスクの配列を作成
const tasks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];

// キューを初期化（同時実行数 = 1）
const queue = async.queue((task, executed) => {
  console.log("現在タスク " + task + " を処理中");

  setTimeout(()=>{
    // 残りのタスク数を取得
    const tasksRemaining = queue.length();
    executed(null, {task, tasksRemaining});
  }, 1000);

}, 1); // concurrency値 = 1

// 要素がないため、最初はキューはアイドル状態
console.log(`Queue Started ? ${queue.started}`)

// タスクリストから各タスクをキューへ追加
tasks.forEach((task)=>{

  // タスク5は先頭に挿入して優先的に実行
  if(task == 5){
    queue.unshift(task, (error, {task, tasksRemaining})=>{
      if(error){
        console.log(`タスク ${task} の処理中にエラーが発生しました`);
      }else {
        console.log(`タスク ${task} の処理完了。残り ${tasksRemaining} タスク`);
      }
    })
  // タスク5以外はすべて末尾に追加
  } else {
    queue.push(task, (error, {task, tasksRemaining})=>{
      if(error){
        console.log(`タスク ${task} の処理中にエラーが発生しました`);
      }else {
        console.log(`タスク ${task} の処理完了。残り ${tasksRemaining} タスク`);
      }
    })
  }
});

// すべてのタスクの処理完了時にコールバックを実行
queue.drain(() => {
  console.log('すべての項目の処理が正常に完了しました！');
})

// タスク追加後、キューが開始されたかどうかを確認
console.log(`Queue Started ? ${queue.started}`)

出力結果

C:\home\node>> node asyncQueue.js
Queue Started ? False
Queue Started ? True
Currently Busy Processing Task 5
Finished processing task 5. 9 tasks remaining
Currently Busy Processing Task 1
Finished processing task 1. 8 tasks remaining
Currently Busy Processing Task 2
Finished processing task 2. 7 tasks remaining
Currently Busy Processing Task 3
Finished processing task 3. 6 tasks remaining
Currently Busy Processing Task 4
Finished processing task 4. 5 tasks remaining
Currently Busy Processing Task 6
Finished processing task 6. 4 tasks remaining
Currently Busy Processing Task 7
Finished processing task 7. 3 tasks remaining
Currently Busy Processing Task 8
Finished processing task 8. 2 tasks remaining
Currently Busy Processing Task 9
Finished processing task 9. 1 tasks remaining
Currently Busy Processing Task 10
Finished processing task 10. 0 tasks remaining
All items are succesfully processed !

まとめ

この出力結果から分かるように、unshift()で先頭に追加されたタスク5が最初に処理され、その後push()で追加されたタスク1〜10が順番に実行されています。同時実行数（concurrency）を変更すれば、一度に並行処理できるタスクの数も調整可能です。

async.queue()を活用することで、大量の非同期タスクを効率的に制御でき、APIへの連続リクエストやファイルの一括処理など、さまざまな場面で役立ちます。ぜひ実際のプロジェクトで試してみてください。