FastAPIとCelery (2026): BackgroundTasks vs 分散キュー

FastAPI で非同期 API を構築するすべての開発者は、最終的に同じ岐路に立たされます。それは、エンドポイントが標準の HTTP リクエストで許容されるよりも長い時間かかる処理を実行する必要がある場合です。

オンボーディングメールの送信、Stripe ウェブフック同期のディスパッチ、アップロードされた動画のトランスコード、ドキュメントチャンクのベクトルデータベースへのインデックス作成、50 ページの財務 PDF の生成などが必要になるかもしれません。

FastAPI の公式ドキュメントには、魅力的なほどシンプルな組み込み機能が紹介されています。

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_welcome_email(email: str):
    # Sends email...
    pass

@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email)
    return {"status": "accepted", "message": "Verification email queued"}

これは魔法のように見えます。追加のインフラストラクチャは不要で、Redis ブローカーも、Celery ワーカーデーモンも、オーケストレーションする Docker Compose サービスもありません。関数を 1 回呼び出すだけで、クライアントは瞬時に 200 OK または 202 Accepted のレスポンスを受け取り、関数はバックグラウンドで実行されます。

しかし、本番環境にデプロイすると状況は一変します。

Kubernetes ポッドは OOM (Out-Of-Memory) キル制限に達し始めます。誰かがレポートを生成するたびに、Uvicorn のイベントループが 4 秒間フリーズします。ローリングデプロイメントは、300 の処理中のタスクをエラーの痕跡を残さずに静かにパージします。そして、SendGrid が一時的な 502 Bad Gateway を返した場合、メールは再試行されることなく永久に失われます。

このガイドは、FastAPI BackgroundTasks と Celery のアーキテクチャに関する詳細な分析です。両者の内部メカニズムを解剖し、10,000 タスクのバースト下でのメモリとレイテンシーのベンチマークを分析し、致命的な本番環境での落とし穴を強調し、2026 年に向けた実証済みの意思決定フレームワークを提供します。

高レベルアーキテクチャ: インプロセス vs 分散

BackgroundTasks と Celery の根本的な違いは、構文やライブラリに関するものではありません。それは、障害境界、リソース分離、状態永続性というアーキテクチャ上の問題です。

1. FastAPI BackgroundTasks: インプロセスの一時的な実行

FastAPI は BackgroundTasks を Starlette (starlette.background.BackgroundTasks) から直接継承しています。background_tasks.add_task() を呼び出すと、Starlette は呼び出し可能オブジェクトとその引数を、Response オブジェクトにアタッチされたシンプルなインメモリ Python リストに追加します。

# Starlette internal implementation pattern
class BackgroundTasks:
    def __init__(self, tasks: list[BackgroundTask] | None = None):
        self.tasks = list(tasks) if tasks else []

    def add_task(self, func: typing.Callable, *args, **kwargs) -> None:
        self.tasks.append(BackgroundTask(func, *args, **kwargs))

    async def __call__(self) -> None:
        for task in self.tasks:
            await task()

ルートハンドラーがレスポンスを返すと、Starlette は HTTP ヘッダーとボディを ASGI ソケット経由でクライアントに送信します。ソケット送信が完了した後にのみ、Starlette は self.tasks を反復処理し、同じ Uvicorn ワーカープロセス内でそれぞれを順次実行します。

• 永続性なし: タスクキューはウェブワーカープロセスの RAM 内に存在します。
• 共有コンピューティング: バックグラウンドタスクは、API サーバーに割り当てられたまったく同じ CPU コアと RAM を消費します。
• 結合されたライフサイクル: ウェブワーカーが終了、クラッシュ、または Kubernetes や Systemd によって再起動された場合、保留中および処理中のすべてのタスクは即座に停止します。

2. Celery: 分散型で疎結合なタスクキュー

Celery は、外部の永続的なメッセージブローカー（Redis や RabbitMQ など）を介して、タスクプロデューサーとタスクコンシューマーを疎結合にします。

1. プロデューサー (FastAPI): タスク引数（通常は JSON）をシリアライズし、AMQP メッセージをブローカーキュー（tasks.send_welcome_email）にパブリッシュします。HTTP リクエストは 3 ミリ秒未満で完了します。
2. ブローカー (Redis / RabbitMQ): タスクメッセージをメモリまたはディスクに永続的に保存します。ウェブおよびワーカーポッドがすべて再起動されても、タスクメッセージは保持されます。
3. コンシューマー (Celery ワーカー): 専用のマシンまたは個別のコンテナで実行される独立したワーカープロセスは、ブローカーを継続的にポーリングし、ジョブを実行し、指数バックオフで再試行を処理し、実行結果を結果バックエンドに書き戻します。

アーキテクチャ比較マトリックス

BackgroundTasks の 4 つの生産上の落とし穴

BackgroundTasks が大規模で失敗する理由を理解するには、Python ASGI ライフタイムのランタイム動作を調べる必要があります。

落とし穴 1: イベントループの飢餓状態

Python の非同期ウェブ開発における最も危険なバグの 1 つは、意図せずにイベントループをブロックしてしまうことです。FastAPI は 2 種類の関数を扱います。

1. async def: メインの asyncio イベントループで直接実行されます。
2. def (標準同期): Starlette によって anyio ワーカーのスレッドプール (to_thread.run_sync) にオフロードされます。

background_tasks.add_task() に関数を渡すと、Starlette はまったく同じロジックを適用します。

# If your task is declared as `async def`
async def sync_stripe_data(user_id: str):
    # TRAP: If you call a blocking library inside an async function:
    import requests # SYNCHRONOUS BLOCKING HTTP CLIENT!
    response = requests.get(f"https://api.stripe.com/v1/customers/{user_id}")

sync_stripe_data は async def で定義されているため、FastAPI はそれを単一スレッドの asyncio イベントループで直接実行します。requests.get() が Stripe のネットワーク応答を 1,500 ミリ秒待機してブロックすると、Uvicorn ワーカープロセス全体がフリーズします。

この 1,500 ミリ秒の間、そのワーカーは新しい受信 TCP 接続を受け入れることができず、SSL ハンドシェイクを処理できず、ヘルスチェックを提供できません。このフリーズ中に Kubernetes が GET /healthz プローブを送信すると、プローブは失敗します。3 回のプローブ失敗後、Kubernetes はポッドを終了します。

落とし穴 2: 耐久性ゼロとローリングデプロイメントの惨事

最新のクラウドインフラストラクチャは継続的デプロイメントに依存しています。Kubernetes、AWS ECS、Fly.io は常に新しいコンテナイメージを展開し、CPU 負荷に基づいてポッドをスケールアップおよびスケールダウンします。

Kubernetes がポッドを終了すると、SIGTERM シグナルを発行し、猶予期間（デフォルト 30 秒）を待ってから SIGKILL を送信します。

次の一連のイベントを想像してみてください。

1. 14:00:00 に、50 人のユーザーが注文リクエストを送信します。
2. FastAPI は 50 人すべてのユーザーに 202 Accepted を返し、process_order_billing を BackgroundTasks にキューイングします。
3. 14:00:01 に、CI/CD パイプラインが新しい本番デプロイメントをトリガーします。Kubernetes は既存のポッドに SIGTERM を発行します。
4. Uvicorn はすぐに新しい接続の受け入れを停止し、シャットダウンを開始します。
5. 実行中のタスクは突然中断されます。Starlette の self.tasks リストにあり、まだ開始されていなかったタスクは、メモリから永久に消去されます。
6. クライアントは注文が処理中であると信じています。データベースには支払いの記録がありません。サポートチームには失敗の記録がありません。

対照的に、Celery では、タスクは Redis または RabbitMQ に安全に存在します。ワーカーが SIGTERM を受信すると、現在のタスクを完了するか、task_reject_on_worker_lost = True と task_acks_late = True で設定されている場合は、メッセージをブローカーに返して、別のワーカーがすぐにそれを取得します。

落とし穴 3: メモリリークと無制限のキュー増大

FastAPI では、BackgroundTasks にはバックプレッシャーメカニズムがありません。API が 1 分あたり 5,000 件のリクエストを受け取り、各リクエストが 200 ミリ秒の CPU 時間を要する画像リサイズタスクをキューに入れる場合、タスク生成レート (5,000/分) はワーカーのシングルコア処理能力 (300/分) をはるかに超えます。

RAM 内の self.tasks リストは単調に増加します。Python の辞書、画像バッファ、クロージャがヒープに蓄積されます。数分以内に、コンテナは cgroup メモリクォータを超過し、Linux カーネルは OOM キラー (exit code 137) をトリガーします。

Celery の場合:

• ブローカーは、設定可能な制限を持つ弾力的なバッファとして機能します。
• タスクはすぐに Redis にフラッシュされるため、ウェブノードは軽量のままです。
• ワーカーの容量は、キューの長さに応じて KEDA (Kubernetes Event-driven Autoscaling) を使用して独立して自動スケーリングできます。

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: celery-worker-scaler
spec:
  scaleTargetRef:
    name: celery-worker
  minReplicaCount: 2
  maxReplicaCount: 50
  triggers:
  - type: redis
    metadata:
      address: redis-service:6379
      listName: celery
      listLength: "50"

落とし穴 4: 再試行と指数バックオフの欠如

ネットワーク呼び出しは失敗します。外部 API はレート制限をかけます。データベースは一時的なロック競合を経験します。

FastAPI の BackgroundTasks 内で堅牢な再試行ロジックを記述するには、車輪の再発明が必要です。

# Re-inventing what Celery already provides out of the box
async def fragile_background_task(user_id: int):
    retries = 3
    delay = 2
    for attempt in range(retries):
        try:
            await push_telemetry(user_id)
            break
        except ExternalAPIError as exc:
            if attempt == retries - 1:
                logger.critical("Telemetry failed permanently: %s", exc)
                # Where do you store the failed payload? Nowhere!
                raise
            await asyncio.sleep(delay)
            delay *= 2 # Exponential backoff

asyncio.sleep(delay) の実行中にサーバーがクラッシュすると、再試行チェーンは消滅します。

Celery は本番環境レベルの再試行をネイティブに提供します。

@celery_app.task(
    bind=True,
    autoretry_for=(ExternalAPIError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=5,
    acks_late=True
)
def robust_celery_task(self, user_id: int):
    push_telemetry(user_id)

Celery は次の再試行時間を計算し、サンダリング・ハード問題を防ぐためにランダムなジッターを適用し、ETA カウントダウン付きでタスクをブローカーに再キューイングします。

コード実装: 並列比較

BackgroundTasks から Celery へ移行する際の、クリーンな関心事の分離を見てみましょう。

# main.py
import time
from fastapi import FastAPI, BackgroundTasks, status
from pydantic import BaseModel, EmailStr

app = FastAPI(title="Notification Gateway")

class WelcomeEmailRequest(BaseModel):
    user_id: int
    email: EmailStr

def dispatch_email_notification(email: str, user_id: int):
    """
    Synchronous function executed in Starlette threadpool.
    Acceptable ONLY for trivial, non-critical notifications.
    """
    # Simulate network latency to SMTP server
    time.sleep(1.2)
    print(f"Sent welcome email to user {user_id} at {email}")

@app.post("/users/welcome", status_code=status.HTTP_202_ACCEPTED)
def send_welcome(payload: WelcomeEmailRequest, tasks: BackgroundTasks):
    tasks.add_task(dispatch_email_notification, payload.email, payload.user_id)
    return {"status": "queued", "user_id": payload.user_id}

# tasks.py (Worker Service)
import os
import time
from celery import Celery
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")

celery_app = Celery(
    "worker",
    broker=REDIS_URL,
    backend=REDIS_URL
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    # Crucial enterprise reliability flags:
    task_acks_late=True,                 # Ack only after execution finishes
    task_reject_on_worker_lost=True,     # Re-queue if worker dies mid-task
    worker_prefetch_multiplier=1,        # Fair task distribution across workers
    worker_max_tasks_per_child=100       # Prevent C-level memory fragmentation
)

@celery_app.task(
    bind=True,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_kwargs={"max_retries": 5}
)
def process_report_generation(self, user_id: int, report_type: str):
    logger.info(f"Generating {report_type} for user {user_id}")
    time.sleep(3.5) # Heavy compute simulation
    return {"user_id": user_id, "status": "completed", "url": "https://s3.amazonaws.com/..."}

# api.py (FastAPI Producer)
from fastapi import FastAPI, status, HTTPException
from pydantic import BaseModel
from tasks import process_report_generation

app = FastAPI(title="Report Processing Gateway")

class ReportRequest(BaseModel):
    user_id: int
    report_type: str

class TaskStatusResponse(BaseModel):
    task_id: str
    status: str

@app.post("/reports/generate", status_code=status.HTTP_202_ACCEPTED, response_model=TaskStatusResponse)
def trigger_report(payload: ReportRequest):
    # .delay() publishes task to Redis in ~2-4ms
    async_result = process_report_generation.delay(payload.user_id, payload.report_type)
    return TaskStatusResponse(task_id=async_result.id, status="PENDING")

@app.get("/reports/status/{task_id}")
def get_report_status(task_id: str):
    result = process_report_generation.AsyncResult(task_id)
    if result.state == "PENDING":
        return {"task_id": task_id, "state": result.state}
    elif result.state == "SUCCESS":
        return {"task_id": task_id, "state": result.state, "result": result.result}
    elif result.state == "FAILURE":
        return {"task_id": task_id, "state": result.state, "error": str(result.info)}
    return {"task_id": task_id, "state": result.state}

パフォーマンスと負荷ベンチマーク: 10,000 タスクのバースト

両方のアーキテクチャの実際の運用コストを定量化するために、標準的な 2-vCPU、2GB RAM コンテナに対して、60 秒以内に 10,000 回のタスク呼び出しというトラフィック急増をシミュレートする合成負荷テストを実施しました。

各タスクは、平均ネットワークレイテンシーが 250 ミリ秒の外部サービスを呼び出す I/O ペイロードをシミュレートします。

Hardware: 2 vCPU, 2GB RAM, Ubuntu 24.04 LTS
ASGI Server: Uvicorn with 2 worker processes
Broker: Redis 7.2 Alpine (in-memory)
Load Generator: k6 with 200 virtual users (VUs)

ベンチマーク結果

ベンチマーク分析

1. 取り込みレイテンシー: FastAPI は、初期 HTTP レスポンスを返すのが速い（約 1.2 ミリ秒 vs 約 3.4 ミリ秒）。これは、RAM 内の Python リストにオブジェクトを追加するのにネットワークソケット操作が不要であるのに対し、Celery は JSON をシリアライズして Redis に LPUSH を実行する必要があるためです。
2. システム安定性: 継続的な負荷の下では、FastAPI のインプロセスキューは大規模なメモリ肥大化（88 MB から 820 MB へ）を引き起こします。ワーカーのスレッドが受信 HTTP 接続と CPU タイムスライスを競合するため、HTTP レスポンスのレイテンシーは著しく低下します。
3. キューのドレイン: Celery は 10,000 個のメッセージすべてを Redis に瞬時にバッファリングします。ウェブサーバーは落ち着いて応答性を保ち、CPU 使用率は 10% 未満で動作します。一方、独立した Celery ワーカープールは、独自の持続可能なペースでキューを着実にドレインします。

2026 年に向けた必須の Celery 設定

本番環境で Celery をデプロイする場合、デフォルト設定は避けてください。Celery は、そのままでは高スループットで重要度の低いタスク向けに調整されており、データ損失やメモリリークにつながる可能性があります。

これらのエンタープライズ設定フラグを適用してください。

# celery_config.py
broker_url = "redis://redis-cluster:6379/0"
result_backend = "redis://redis-cluster:6379/1"

# 1. Prevent memory leaks from third-party libraries (e.g., NumPy, Pandas, PyTorch)
worker_max_tasks_per_child = 500  # Worker process restarts cleanly after 500 tasks
worker_max_memory_per_child = 250000  # 250MB limit before recycling

# 2. Guarantee At-Least-Once Delivery
task_acks_late = True
task_reject_on_worker_lost = True

# 3. Prevent worker starvation from greedy prefetching
# By default, Celery prefetches 4 tasks per worker thread. If one task takes 10 minutes,
# the other 3 prefetched tasks sit blocked while idle workers have empty queues.
worker_prefetch_multiplier = 1

# 4. Result cleanup to prevent Redis memory exhaustion
result_expires = 86400  # 24 hours TTL for task results

# 5. Broker connection resiliency
broker_connection_retry_on_startup = True
broker_transport_options = {
    "visibility_timeout": 43200,  # 12 hours (must exceed longest task duration)
}

いつ何を使うか: エンジニアリング意思決定フレームワーク

アーキテクチャ要件に合った適切なツールを選択するために、この意思決定リファレンスを使用してください。

最新の代替案: Celery だけが唯一の選択肢なのか？

Celery は Python における議論の余地のないエンタープライズ標準であり続けていますが、いくつかの現代的な代替案が大きな注目を集めています。

知識の確認

よくある質問

結論

FastAPI の BackgroundTasks は、軽量で重要度の低いインプロセスでの副作用（ローカルメモリカウンターの更新、ディスク上の一時アップロードファイルのクリーンアップ、再デプロイ時に時折データ損失が許容されるファイア＆フォーゲットのテレメトリーピングの送信など）に優れたツールです。

しかし、タスクが次のいずれかに該当する場合、

1. 重要なビジネスロジック（支払い、ユーザーメール、注文処理）を実行する、
2. 保証された実行と指数バックオフによる自動再試行を要求する、
3. 重い CPU 計算や長時間実行されるワークフローを実行する、
4. または API サーバーとは独立した水平スケーリングを必要とする、

Celery + Redis は過剰ではありません。それはアーキテクチャ上の必然です。 ウェブ層と非同期計算層を分離することで、API は高速に動作し、メモリフットプリントは予測可能に保たれ、ユーザーリクエストは本番環境で発生するあらゆる事態を乗り越えることができます。

こちらもおすすめ

• FastAPI vs Litestar (2026): パフォーマンス、ベンチマーク、切り替え時期
• 本番環境での非同期 Python メモリリークのプロファイリング
• 本番環境での SQLite: WAL モード、高並行性、実証済みの PRAGMA
• Polars vs Pandas: パフォーマンスとメモリのベンチマーク
• 2026 年に期待できる Python ライブラリ