BullMQ - Node.jsの最強ジョブキューを使いこなす

はじめに

バックエンド開発をしていると、「重い処理を非同期で実行したい」という場面に必ず遭遇する。メール送信、画像処理、外部API連携など、ユーザーを待たせたくない処理は山ほどあるんですよね。

そんなときに活躍するのがジョブキューという仕組みで、Node.js界隈ではBullMQが個人的には一択だと思っています。

BullMQは「最速で最も信頼性の高いRedisベースの分散キュー」を謳っているライブラリで、GitHub スター数は8,000以上、月間ダウンロード数は830万を超える実績があります。SlackやTwilioといった企業でも採用されているので、信頼性は折り紙付きという話。

BullMQの特徴・メリット

正直なところ、Node.jsのジョブキューライブラリは他にもあるんですが、BullMQを推す理由を挙げておきます。

堅牢性がすごい

アトミック操作: Redisのトランザクションを活用して、ジョブの状態管理が確実
自動リトライ: ワーカーがクラッシュしても、ジョブが消えない
Exactly-once semantics: 基本的に1回だけ実行、最悪でもat-least-once保証

柔軟なジョブ制御

遅延実行: 「5分後に実行」みたいな指定が簡単
優先度: 急ぎのジョブを先に処理
繰り返し実行: cronライクなスケジューリング
レート制限: 外部APIの制限に合わせた実行制御

スケーラビリティ

水平スケーリング: ワーカーを増やすだけで処理能力アップ
並行処理: ワーカーごとに同時実行数を設定可能
サンドボックス: 別プロセスで安全に実行

開発体験

TypeScriptネイティブ: 型定義が最初から完璧
イベントドリブン: ジョブの完了や失敗をリアルタイムで監視
親子関係: ジョブ間の依存関係を表現可能

インストール方法

前提としてRedisが必要なので、まずはRedisを用意してください。ローカル開発ならDockerが楽ですね。

# Redisの起動（Docker）
docker run -d --name redis -p 6379:6379 redis

# BullMQのインストール
npm install bullmq
# または
yarn add bullmq

これだけで準備完了です。シンプルでいいですよね。

基本的な使い方

キューの作成とジョブの追加

まずはキューを作成して、ジョブを追加してみます。

import { Queue } from 'bullmq';

// キューの作成
const emailQueue = new Queue('email', {
  connection: {
    host: 'localhost',
    port: 6379,
  },
});

// ジョブの追加
await emailQueue.add('welcome-email', {
  to: 'user@example.com',
  subject: 'ようこそ！',
  body: '登録ありがとうございます。',
});

console.log('ジョブを追加しました');

addメソッドの第1引数がジョブ名、第2引数がジョブに渡すデータです。これだけで、Redisにジョブが保存されます。

ワーカーの作成

次に、ジョブを処理するワーカーを作成します。

import { Worker } from 'bullmq';

const worker = new Worker(
  'email',
  async (job) => {
    console.log(`Processing job ${job.id}`);
    console.log(`Sending email to: ${job.data.to}`);

    // 実際のメール送信処理
    await sendEmail(job.data);

    return { sent: true };
  },
  {
    connection: {
      host: 'localhost',
      port: 6379,
    },
    concurrency: 5, // 同時に5つまで処理
  }
);

worker.on('completed', (job) => {
  console.log(`Job ${job.id} completed`);
});

worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed:`, err.message);
});

concurrencyオプションで同時実行数を指定できるのが便利。外部APIの制限がある場合は1にしておくと安全です。

遅延実行とリトライ

実務で特によく使うオプションを紹介します。

// 5分後に実行
await emailQueue.add(
  'reminder',
  { userId: 123 },
  { delay: 5 * 60 * 1000 }
);

// 失敗時のリトライ設定
await emailQueue.add(
  'important-email',
  { to: 'vip@example.com' },
  {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000,
    },
  }
);

backoffをexponentialにすると、1秒→2秒→4秒と間隔を空けてリトライしてくれる。外部サービスが一時的に落ちてるときに重宝するんですよね。

繰り返しジョブ（Cron）

定期実行したい処理はこんな感じで書けます。

// 毎日朝9時に実行
await emailQueue.add(
  'daily-report',
  { type: 'daily' },
  {
    repeat: {
      pattern: '0 9 * * *', // cron形式
      tz: 'Asia/Tokyo',
    },
  }
);

タイムゾーン指定ができるのが地味にありがたい。

実践的なユースケース

メール送信キュー

これが一番よくあるパターンだと思います。

import { Queue, Worker } from 'bullmq';
import { z } from 'zod';

// 型定義
const EmailJobSchema = z.object({
  to: z.string().email(),
  subject: z.string(),
  template: z.string(),
  data: z.record(z.unknown()),
});

type EmailJob = z.infer<typeof EmailJobSchema>;

// キューの作成
const emailQueue = new Queue<EmailJob>('email');

// プロデューサー（ジョブを追加する側）
export async function queueEmail(email: EmailJob) {
  const validated = EmailJobSchema.parse(email);
  return emailQueue.add('send', validated, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: 100, // 完了後100件まで保持
    removeOnFail: 500,     // 失敗後500件まで保持
  });
}

// コンシューマー（ジョブを処理する側）
const worker = new Worker<EmailJob>(
  'email',
  async (job) => {
    const html = await renderTemplate(job.data.template, job.data.data);
    await sendgrid.send({
      to: job.data.to,
      subject: job.data.subject,
      html,
    });
  },
  { concurrency: 10 }
);

画像処理パイプライン

重い処理の代表格。親子ジョブの機能を使うと、処理の流れを表現できます。

import { FlowProducer } from 'bullmq';

const flowProducer = new FlowProducer();

// 親子関係のあるジョブを一括追加
await flowProducer.add({
  name: 'finalize',
  queueName: 'image',
  data: { outputPath: '/uploads/processed/' },
  children: [
    {
      name: 'resize',
      queueName: 'image',
      data: { size: 'thumbnail' },
    },
    {
      name: 'resize',
      queueName: 'image',
      data: { size: 'medium' },
    },
    {
      name: 'optimize',
      queueName: 'image',
      data: { quality: 80 },
    },
  ],
});