第6章 Celery と Redis による非同期タスク基盤

Celery は、Python で最も広く利用されている実績豊富な分散タスクキューフレームワークです。

本章では、Celery のコアアーキテクチャ、Redis を使ったメッセージングの仕組み、タスクの定義方法、ワーカーの起動、および障害に強いリトライ設計を学びます。

6.1 Celery の基本アーキテクチャ

Celery は主に以下の 4 つの要素で構成されます。

クライアント（Producer）: タスクの実行を依頼する側です（FastAPI アプリケーションなど）。
メッセージブローカー（Broker）: タスクの指示書（メッセージ）を安全にキューイングして蓄積する中間システムです（Redis や RabbitMQ）。
ワーカー（Consumer）: ブローカーからタスクを取り出してバックグラウンドで実際に処理を実行する Python プロセスです。
結果バックエンド（Result Backend）: タスクの実行結果や成否、進捗状態を保存しておくストレージです（Redis やデータベース）。

graph LR
    Client["FastAPI (Client)"] -->|"1. タスク発行"| Broker[("Redis (Broker)")]
    Broker -->|"2. タスク取得"| Worker["Celery Worker"]
    Worker -->|"3. 結果保存"| Backend[("Redis (Backend)")]
    Client -.->|"4. 結果問い合わせ"| Backend

6.2 Redis のセットアップ

Redis は、軽量・高速でセットアップが容易なため、Celery のメッセージブローカーおよび結果バックエンドとして最適です。

ローカル開発環境では、Docker を使用して数秒で起動できます。

docker run -d --name redis-server -p 6379:6379 redis:alpine

6.3 Celery アプリケーションの初期化とタスク定義

Celery インスタンスを作成し、タスクとなる関数に @celery_app.task デコレータを付与します。

以下は、celery_tasks.py というファイルにタスクを定義する例です。

import time
from celery import Celery

# Redis への接続 URL を指定
REDIS_URL = "redis://localhost:6379/0"

# Celery アプリケーションの初期化
celery_app = Celery(
    "tasks",
    broker=REDIS_URL,
    backend=REDIS_URL,
)

# シリアライザやタイムゾーンの設定
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Asia/Tokyo",
    enable_utc=True,
)


# バックグラウンドで実行したいタスク
@celery_app.task(bind=True)
def long_running_task(self, x: int, y: int) -> int:
    print(f"タスク開始: ID={self.request.id}")
    time.sleep(5)  # 5秒かかる重い計算をシミュレート
    result = x + y
    print(f"タスク完了: 結果={result}")
    return result

bind=True を指定すると、関数の第 1 引数にタスク自身のインスタンス（self）が渡され、タスク ID の参照や進捗の更新が可能になります。

6.4 タスクの呼び出し（delay と apply_async）

定義したタスクを呼び出す方法は 2 種類あります。

task.delay(*args, **kwargs): 最もシンプルな呼び出し方です。引数を渡して即座にブローカーに投入します。
task.apply_async(args, kwargs, ...): 詳細な実行オプションを指定できます。カウントダウン（例: 60 秒後に実行）、優先度、有効期限、特定キューへのルーティングなどを指定できます。

# 1. シンプルな即時投入
result = long_running_task.delay(10, 20)
print(f"発行されたタスクID: {result.id}")

# 2. 30秒後に実行する遅延投入
result_delayed = long_running_task.apply_async(args=[10, 20], countdown=30)

どちらのメソッドも、呼び出した瞬間に AsyncResult オブジェクトを返します。この時点ではタスクはワーカーで処理中であり、待たされることはありません。

6.5 ワーカープロセスの起動

定義したタスクを実行するためには、ターミナルでワーカープロセスを起動します。

# celery_tasks.py の celery_app を指定してワーカーを起動
celery -A celery_tasks.celery_app worker --loglevel=info --concurrency=4

-A celery_tasks.celery_app: アプリケーションの場所を指定します。
--loglevel=info: ログの詳細度を指定します。
--concurrency=4: 同時に並行処理するタスク数（ワーカープロセス数）を指定します。

6.6 タスクの状態確認と自動リトライ

発行されたタスク ID を使って、結果バックエンドから現在の状態や戻り値を取得できます。

from celery.result import AsyncResult
from celery_tasks import celery_app

res = AsyncResult("発行されたタスクID", app=celery_app)

print("状態:", res.status)  # PENDING, STARTED, SUCCESS, FAILURE など
if res.ready():
    print("結果:", res.result)

また、ネットワークエラーや一時的な障害に備えて、自動リトライを宣言的に設定できます。

import httpx


@celery_app.task(
    bind=True,
    autoretry_for=(httpx.RequestError,),  # 通信エラー時に自動再試行
    retry_kwargs={"max_retries": 3, "countdown": 5},  # 最大3回、5秒間隔
    retry_backoff=True,  # 失敗するごとに間隔を延ばす（指数バックオフ）
)
def fetch_remote_webhook(self, url: str):
    res = httpx.post(url, json={"ping": "pong"}, timeout=5.0)
    res.raise_for_status()
    return res.status_code