【Node.js】メッセージキューを使った非同期ジョブ処理の実装｜Bull×Redis入門

Node.jsで大量のタスクを効率よく処理したいとき、非同期ジョブキューの活用が有効です。特に、重たい処理をバックグラウンドで並列実行しつつ、安定性・再試行性・監視性も担保したい場合には、BullとRedisを組み合わせた構成が強力です。

この記事では、メッセージキューの概念から、Bullを使ったジョブキュー処理の実装までを実例付きで解説します。

なぜメッセージキューが必要なのか？
Bullとは？
準備：インストールとRedisの起動
基本的なキューの作成とジョブ登録
リトライ・遅延・繰り返しなどの応用設定
監視ツール：Bull Boardの導入
まとめ｜Node.jsのスケーラブルなジョブ設計にBullを

なぜメッセージキューが必要なのか？

Web APIやバッチ処理では、以下のようなケースが存在します：

大量のメール送信・PDF生成・画像圧縮などの重い処理を逐次実行するとAPIレスポンスが遅くなる
同時に大量アクセスが来ると、Node.jsのイベントループがブロックされる可能性がある
処理が途中で失敗した場合の再実行やリトライ制御が面倒

こうした問題に対して、メッセージキュー（ジョブキュー）は、タスクの登録と非同期実行を分離し、システム全体のスケーラビリティと保守性を高める手段となります。

Bullとは？

Bullは、Node.js製の人気キューライブラリで、Redisをバックエンドに使います。主な特徴は以下の通りです：

ジョブの優先度設定・遅延実行・繰り返し処理
失敗時のリトライ設定
同時並行実行（コンカレンシー）
ダッシュボードによる監視（Bull Board）

準備：インストールとRedisの起動

npm install bull

Redisが必要なので、まだ起動していない場合は以下のようにDockerで立ち上げることもできます：

docker run -d -p 6379:6379 redis

基本的なキューの作成とジョブ登録

まずは、ジョブを登録する側のコードです。

// queue/addJob.js
const Queue = require('bull');
const myQueue = new Queue('myJobQueue', 'redis://127.0.0.1:6379');

// ジョブ登録
myQueue.add({ message: 'こんにちは！ジョブキュー処理開始' });

次に、ジョブを処理するワーカー側のコードです。

// worker/jobWorker.js
const Queue = require('bull');
const myQueue = new Queue('myJobQueue', 'redis://127.0.0.1:6379');

myQueue.process(async (job) => {
  console.log('ジョブを処理中:', job.data.message);
  // 時間のかかる処理をここで実行
});

この構成で、ジョブ追加と処理が完全に非同期で行えるようになります。

リトライ・遅延・繰り返しなどの応用設定

myQueue.add(
  { userId: 123 },
  {
    attempts: 3, // 最大3回リトライ
    delay: 5000, // 5秒後に実行
    repeat: { cron: '0 * * * *' } // 毎時0分に繰り返し
  }
);

監視ツール：Bull Boardの導入

npm install @bull-board/express

Expressと統合してWeb上でジョブの状況を監視できます。

const express = require('express');
const { ExpressAdapter } = require('@bull-board/express');
const { createBullBoard } = require('@bull-board/api');

const Queue = require('bull');
const myQueue = new Queue('myJobQueue');

const serverAdapter = new ExpressAdapter();
const { addQueue, removeQueue, setQueues } = createBullBoard({
  queues: [new (require('@bull-board/api').BullAdapter)(myQueue)],
  serverAdapter,
});

const app = express();
serverAdapter.setBasePath('/admin/queues');
app.use('/admin/queues', serverAdapter.getRouter());

app.listen(3000, () => console.log('Bull Board起動 http://localhost:3000/admin/queues'));