background-jobs-designer

バックグラウンドジョブ設計

リトライと監視を備えた、信頼性の高いバックグラウンドジョブ処理を設計します。

キューの統合

BullMQ（Node.js）：

import { Queue, Worker } from "bullmq";

const emailQueue = new Queue("email", {
  connection: { host: "localhost", port: 6379 },
});

// Add job
await emailQueue.add(
  "send-welcome",
  {
    userId: "123",
    email: "[email protected]",
  },
  {
    attempts: 3,
    backoff: { type: "exponential", delay: 2000 },
  }
);

Celery（Python）：

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task(bind=True, max_retries=3)
def send_email(self, user_id, email):
    try:
        # Send email
        pass
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

ジョブの定義

export interface Job {
  id: string;
  type: string;
  payload: unknown;
  attempts: number;
  maxAttempts: number;
  createdAt: Date;
  processedAt?: Date;
  failedAt?: Date;
  error?: string;
}

export const JOB_TYPES = {
  SEND_EMAIL: "send-email",
  PROCESS_PAYMENT: "process-payment",
  GENERATE_REPORT: "generate-report",
  SYNC_DATA: "sync-data",
} as const;

リトライ戦略

// Exponential backoff
const RETRY_CONFIG = {
  maxAttempts: 5,
  delays: [
    1000, // 1 second
    5000, // 5 seconds
    30000, // 30 seconds
    300000, // 5 minutes
    1800000, // 30 minutes
  ],
};

// Worker with retry
const worker = new Worker("email", async (job) => {
  try {
    await sendEmail(job.data);
  } catch (error) {
    if (job.attemptsMade < RETRY_CONFIG.maxAttempts) {
      throw error; // Will retry
    }
    await handleFailedJob(job, error);
  }
});

冪等性のあるジョブ

// Track processed jobs
export const processJob = async (job: Job) => {
  // Check if already processed
  const processed = await db.query(
    "SELECT 1 FROM processed_jobs WHERE job_id = $1",
    [job.id]
  );

  if (processed.rows.length > 0) {
    console.log("Job already processed");
    return; // Idempotent
  }

  await db.transaction(async (trx) => {
    // Mark as processed
    await trx("processed_jobs").insert({ job_id: job.id });

    // Do work
    await performWork(job, trx);
  });
};

監視

// Job events
worker.on("completed", (job) => {
  metrics.increment("jobs.completed", { type: job.name });
});

worker.on("failed", (job, err) => {
  metrics.increment("jobs.failed", { type: job.name });
  logger.error("Job failed", { jobId: job.id, error: err });
});

worker.on("stalled", (jobId) => {
  metrics.increment("jobs.stalled");
  logger.warn("Job stalled", { jobId });
});

ベストプラクティス

• ジョブは冪等性を備えるようにします
• リトライには指数バックオフを使用します
• 妥当なタイムアウトを設定します
• キューの深さを監視します
• 失敗したジョブ用にデッドレターキューを用意します
• ジョブの開始と完了をログに記録します
• グレースフルシャットダウンを処理します

出力チェックリスト

• キューのセットアップ（Redis/RabbitMQ）
• ジョブタイプの定義
• バックオフを伴うリトライポリシー
• 冪等性の追跡
• エラー処理
• 監視／メトリクス
• デッドレターキュー
• グレースフルシャットダウン