開発・技術選定

非同期処理をどう入れるか — ジョブキューを導入する判断

· by IT.Skill

メール送信、PDF生成、外部API呼び出し。これらをリクエストの中で処理すると、
ワーカーが専有され、同時アクセスで詰まります。

かといってジョブキューを入れると運用対象が増えます。段階的に考えます。

段階1：そもそも速くできないか

まず「非同期にする」以外の選択肢を検討します。

  • クエリの改善で 3秒 → 0.2秒になるなら、非同期は不要
  • 外部APIの呼び出しをキャッシュできないか
  • 本当にリアルタイムで必要か

構成を増やさずに済むなら、それが最善です。

段階2：スレッドで逃がす（軽い処理）

メール送信程度なら、これで足りることがあります。

import threading

def send_async(subject, body, to):
    threading.Thread(
        target=lambda: send_mail(subject, body, FROM, to),
        daemon=True,
    ).start()

ただし失敗しても誰も気づきません。
再送やエラー通知が必要な処理には使えません。

段階3：DBをキューにする（中規模まで）

専用ミドルウェアを入れずに、DBのテーブルをキューにします。

class Job(models.Model):
    kind = models.CharField(max_length=40)
    payload = models.JSONField()
    status = models.CharField(max_length=10, default="queued")
    attempts = models.IntegerField(default=0)
    run_after = models.DateTimeField(default=timezone.now)
    locked_at = models.DateTimeField(null=True)

取り出しは行ロックで競合を防ぎます。

with transaction.atomic():
    job = (Job.objects.select_for_update(skip_locked=True)
           .filter(status="queued", run_after__lte=timezone.now())
           .first())
    if job:
        job.status = "running"
        job.locked_at = timezone.now()
        job.save()

skip_locked=True が重要です。
他のワーカーが処理中の行を待たずに飛ばすため、並列実行できます。

この方式の利点は、運用対象が増えないことです。
DBのバックアップにジョブも含まれ、管理画面から状態も見えます。

段階4：専用のジョブキュー

次の条件が出てきたら、専用のものを検討します。

  • 秒間数十件以上を捌く
  • 優先度キューが必要
  • 定期実行を細かく管理したい
  • 処理が数分以上かかり、進捗を追いたい

導入するとブローカーとワーカーの監視が増えます。
それに見合う規模かを判断します。

どの方式でも必要なこと

1. リトライ回数の上限

if job.attempts >= 3:
    job.status = "failed"
    notify_admin(job)

2. 放置されたジョブの回収

stale = Job.objects.filter(
    status="running", locked_at__lt=timezone.now() - timedelta(minutes=10))
stale.update(status="queued", locked_at=None)

ワーカーが落ちると running のまま残ります。これが無いと永遠に処理されません。

3. 冪等性

リトライで2回実行されても問題ない作りにします。
メール送信なら「送信済みフラグ」を先に立てるなどの工夫が要ります。

まとめ

  • まず速くできないかを検討する（構成を増やさないのが最善）
  • 中規模まではDBをキューにすれば十分（select_for_update(skip_locked=True)）
  • 専用キューは規模と要件が出てから
  • どの方式でも「上限・回収・冪等性」の3つは必須

関連記事