クイックスタート（5分）
Webリクエストループのブロッキングは、ユーザー体験を著しく損ないます。エンドポイントが40ページのPDFレポートを生成したり、応答の遅い決済ゲートウェイを呼び出したりすると、レスポンスタイムは容易に45msから4秒以上に跳ね上がります。重い処理をバックグラウンドワーカーにオフロードすることで、APIの応答性を高く維持できます。
CeleryとRedisを使えば、この処理の委譲をシンプルに実現できます。まずはDockerでローカルのRedisブローカーを立ち上げましょう。
docker run -d --name redis-broker -p 6379:6379 redis:7-alpine
必要なPythonの依存パッケージをインストールします。
pip install celery redis
tasks.py 内でバックグラウンドタスクを定義します。
from celery import Celery
import time
app = Celery(
"tasks",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/1"
)
@app.task
def process_signup_email(user_email: str) -> str:
time.sleep(2) # リモートSMTPのレイテンシーをシミュレート
return f"{user_email} 宛てにウェルカムメールを送信しました"
ターミナルでワーカープロセスを起動します。
celery -A tasks worker --loglevel=info --concurrency=2
Python REPLから最初のジョブをキューに追加（エンキュー）します。
>>> from tasks import process_signup_email
>>> async_result = process_signup_email.delay("[email protected]")
>>> async_result.id
'2e9564f2-9860-498c-85a7-96a24eb5ef74'
>>> async_result.ready()
False
>>> async_result.get(timeout=5)
'[email protected] 宛てにウェルカムメールを送信しました'
呼び出し元はミリ秒単位でペイロードを渡し、UUIDを受け取って、HTTPスレッドを即座に解放します。
詳細解説
Celeryをスケールさせるには、クライアント、メッセージトランスポート、OSプロセスがワークロードをどのように連携処理しているかを理解する必要があります。
メッセージブローカーとリザルトバックエンドの違い
Celeryはキューイングロジックと実行結果の追跡を分離しています。
- ブローカー（Redis DB 0）: トランスポートキューとして機能します。
delay()を呼び出すと、タスクの引数がJSON文字列にシリアライズされ、Redisのリストにプッシュされます。Webワーカーは処理を委譲後すぐに接続を解除できます。 - リザルトバックエンド（Redis DB 1）: 実行ステータス（
PENDING、SUCCESS、FAILURE）と戻り値を追跡します。トランザクションメールの送信など、副作用（実行結果の値自体を必要としない処理）のみが目的の場合は、ignore_result=Trueを設定してください。バックエンドを無効化すると一時的なRedisキーが完全に排除され、書き込みIOPSを約50%削減できます。
並行処理モデル: prefork と gevent
デフォルトの prefork プールはPythonの multiprocessing モジュールを利用します。各ワーカープロセスが独立したメモリ空間を持つため、preforkは画像圧縮やデータ分析などのCPUバウンドなジョブにおいてGIL（グローバルインタプリタロック）を回避できます。しかし、OSプロセスは軽量ではありません。32個のpreforkワーカーを起動すると、1.5GB〜3GBのRAMを消費することがあります。
外部REST APIのポーリングやWebスクレイピングなど、ワークロードが主にネットワークI/Oである場合は、イベントループに切り替えましょう。単一の gevent ワーカープロセスなら、80MB未満のRAM消費で数百の並行接続を処理できます。
celery -A tasks worker --pool=gevent --concurrency=100 --loglevel=info
タスクの確認応答（ACK）と冪等性
Celeryはデフォルトでメッセージを早期に確認応答（ACK）します。ワーカーがタスクをデキューした瞬間に、ブローカーはそのタスクを完了済みとしてマークします。そのため、処理の途中でワーカープロセスがOOM（メモリ不足）のスパイクによりLinuxに強制終了された場合、そのタスクは跡形もなく消滅してしまいます。
関数の実行が正常に終了するまで確認応答を保留するには、task_acks_late = True を設定します。ただし、これにはトレードオフがあります。処理の途中でノードがクラッシュした場合、Redisはそのジョブを別のワーカーに再配信します。この設定を有効にする前に、一意の冪等性キー（Idempotency Key）を用いるなどして、すべてのデータベース更新や決済API呼び出しが冪等になるよう設計してください。
高度な活用法
Canvasによるパイプラインのオーケストレーション
本番システムでは、複数ステップにわたる処理が頻繁に必要となります。例えば、カード決済を行い、PostgreSQLに請求書を記録し、プッシュ通知を送信するという一連の処理を順番に行うケースです。Celery Canvasの chain や group を使えば、コールバックのネストなしでこれらのパイプラインを簡潔に組み立てられます。
from celery import Celery, chain, group
app = Celery("ecommerce", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")
@app.task
def charge_customer(order_id: int, amount: float):
return {"order_id": order_id, "transaction_id": f"tx_{order_id}", "amount": amount}
@app.task
def generate_invoice(charge_data: dict):
invoice_number = f"INV-{charge_data['order_id']}"
return {**charge_data, "invoice": invoice_number}
@app.task
def notify_customer(invoice_data: dict):
return f"請求書 {invoice_data['invoice']} の通知を送信しました"
@app.task
def sync_inventory_item(item_id: int):
return f"アイテム {item_id} の在庫を減算しました"
# 順次実行パイプライン: パイプ演算子で出力を次のタスクに渡す
workflow = chain(
charge_customer.s(order_id=4021, amount=99.50) |
generate_invoice.s() |
notify_customer.s()
)
workflow.apply_async()
# 並列実行: 独立したジョブを並行して実行
inventory_sync = group(
sync_inventory_item.s(item_id=101),
sync_inventory_item.s(item_id=102),
sync_inventory_item.s(item_id=103)
)
inventory_sync.apply_async()
外部連携のレートリミット
サードパーティAPIには厳格なレート制限が課されていることが多くあります。例えばTwilioでは、通常番号で毎秒10件を超えるSMSリクエストが拒否されます。Celeryはタスク定義レベルでトークンバケットアルゴリズムによるレートリミッターを標準提供しています。
@app.task(rate_limit="10/s", max_retries=3, default_retry_delay=5)
def send_sms_alert(phone_number: str, message: str):
try:
# 外部SMSプロバイダーへの送信処理
pass
except ExternalServiceError as exc:
# 指数バックオフによるリトライ: 5秒、10秒、20秒
countdown = 5 * (2 ** send_sms_alert.request.retries)
raise send_sms_alert.retry(exc=exc, countdown=countdown)
注意点として、Celeryの rate_limit はクラスタ全体でグローバルに適用されるのではなく、ワーカーインスタンスごとに適用されます。もし rate_limit="10/s" を設定したワーカーコンテナを10台運用している場合、クラスタ全体ではプロバイダーに対して最大で毎秒100件のリクエストが送信される可能性があります。クラスタ全体で厳密なグローバル制限を担保したい場合は、Redisを用いたアトミックなトークンバケットや専用のリバースプロキシを導入してください。
実践的なTips
1. 優先度ごとにキューを分離する
即時性が求められる認証コードと、処理に2時間かかる巨大なレポート出力を同じデフォルトキューに投入するのは障害の元です。突発的に5,000件のPDF出力が発生するとワーカーのスロットがすべて埋まり、パスワードリセットのメールがバックログで待たされることになります。
ルーティング設定でタスクを専用のキューに分割しましょう。
app.conf.task_routes = {
"tasks.send_sms_alert": {"queue": "priority"},
"tasks.generate_invoice": {"queue": "billing"},
"tasks.export_large_csv": {"queue": "batch"},
}
キューごとに専用のワーカープロセスを分離して実行します。
# 高スループットな優先キュー用ワーカー
celery -A tasks worker -Q priority -c 4
# バッチ専用ワーカー（重いジョブによるAPIキューのリソース枯渇を防止）
celery -A tasks worker -Q batch -c 1
2. ワーカーのメモリ上限を設定する
Pandas、ReportLab、Pillowなどのライブラリは、長時間稼働するプロセスでメモリ断片化を引き起こしがちです。子プロセスが徐々に肥大化し、最終的にLinuxのSIGKILLによって強制終了されるリスクがあります。一定量のタスクを処理した後に子プロセスを自動再起動するようにCeleryを設定しましょう。
# 500タスク処理後、またはメモリ使用量が200MBに達した時点で子プロセスを再起動
app.conf.worker_max_tasks_per_child = 500
app.conf.worker_max_memory_per_child = 200000 # 200,000 KB (~195 MB)
3. 処理時間のバラつきが大きい場合はprefetch multiplierを下げる
Celeryはデフォルトでタスクをまとめてプリフェッチ（先読み）します。prefetch_multiplier が4で並行数（concurrency）が4のワーカーは、事前に16個のタスクを確保します。もしそのうち4つのタスクにそれぞれ5分かかる場合、残りの12個のタスクはそのワーカーのローカルバッファで待機状態となり、他の空いているワーカーに割り振られなくなります。
実行時間が長いタスクや変動が大きいジョブでは、プリフェッチを1に設定して、未処理のタスクがRedis上に留まるようにします。
app.conf.worker_prefetch_multiplier = 1
app.conf.task_acks_late = True
4. 明示的な実行タイムアウトを設定する
外部HTTPリクエストの未ハンドルのソケットハングは、ワーカーを無期限にフリーズさせる原因になります。明示的なソフトタイムアウトとハードタイムアウトの制限を設けて、すべてのタスクを保護しましょう。
# ソフトタイムアウト: クリーンアップ処理用にキャッチ可能なSoftTimeLimitExceeded例外を発生させる
# ハードタイムアウト: 10秒後に子プロセスへSIGKILLを送信する
app.conf.task_soft_time_limit = 120
app.conf.task_time_limit = 130
これらの制限を設定することで、サーバーメモリの枯渇を防ぎ、ワーカーのスタックを回避し、本番環境において安定したキューレイテンシーを維持できます。
