時間のかかる処理（外部APIへの通知、画像変換、レポート生成など）をリクエストの中で実行すると、応答が遅くなり、タイムアウトや二重送信の原因になります。Celery を使うと、こうした処理を「タスク」としてキューに積み、別プロセスのワーカーに後から実行させられます。この記事では、Redis をブローカーにした最小構成から、リトライ・タイムアウト・定期実行（Celery Beat）、systemd での常駐化、つまずきやすい点の確認方法までを順に説明します。
Celery の構成要素を先に押さえる
Celery は1つのプログラムではなく、役割の違う部品の組み合わせで動きます。ここを理解しておくと、動かないときにどこを調べればよいかが分かります。
- プロデューサー：タスクを投入する側。Web アプリや CLI スクリプトが
delay()を呼ぶ部分です。 - ブローカー：タスクのメッセージを預かるキュー。Redis か RabbitMQ がよく使われます。
- ワーカー：キューからタスクを取り出して実行するプロセス。
celery workerコマンドで起動します。 - 結果バックエンド：タスクの戻り値や状態を保存する場所（任意）。結果を読まないなら設定しなくても動きます。
- Beat：決まった時刻にタスクを投入するスケジューラー（任意）。
投入した側から見ると「キューに入れた時点で処理は終わっていない」点が重要です。成功したかどうかは結果バックエンドかログで後から確認します。
インストールと最小構成
ここでは Linux サーバー上で、Python の仮想環境と Redis を使う前提で進めます。Celery は安定版の 5 系を想定しています（2026年10月現在）。
python3 -m venv venv
source venv/bin/activate
pip install "celery[redis]" requests
# Redis が起動しているか確認（PONG が返れば OK）
redis-cli pingcelery[redis] のように extras を付けると、Redis 接続に必要なライブラリも一緒に入ります。Windows は Celery の公式サポート対象外とされているため、開発で Windows を使う場合は WSL2 や Docker 上で動かすのが無難です。
ディレクトリ構成は次のようにします。
proj/
├── __init__.py
├── celery_app.py
└── tasks.pycelery_app.py に Celery アプリ本体と設定を書きます。
# proj/celery_app.py
import os
from celery import Celery
app = Celery(
"proj",
broker=os.environ.get("CELERY_BROKER_URL", "redis://localhost:6379/0"),
backend=os.environ.get("CELERY_RESULT_BACKEND", "redis://localhost:6379/1"),
include=["proj.tasks"],
)
app.conf.update(
task_serializer="json",
result_serializer="json",
accept_content=["json"],
timezone="Asia/Tokyo",
enable_utc=True,
result_expires=3600, # 結果は1時間で削除
task_time_limit=300, # 300秒でワーカーが強制終了
task_soft_time_limit=240, # 240秒で SoftTimeLimitExceeded を発生
task_acks_late=True, # 実行完了後にキューから消す
worker_prefetch_multiplier=1, # 長いタスクを1件ずつ取る
broker_connection_retry_on_startup=True,
)ブローカーの URL にパスワードを含める場合は、コードに直接書かず環境変数で渡します。シリアライザーを JSON に限定しているのは、pickle を受け付けると、キューに不正なメッセージを入れられたときに任意のコードを実行される危険があるためです。
タスクを定義して投入する
タスクは @app.task を付けた普通の関数です。外部 Webhook へ通知する例で、失敗時の自動リトライも合わせて設定します。
# proj/tasks.py
import time
from pathlib import Path
import requests
from celery.utils.log import get_task_logger
from .celery_app import app
logger = get_task_logger(__name__)
@app.task(
bind=True,
autoretry_for=(requests.RequestException,),
retry_backoff=True, # 間隔を 1, 2, 4, 8... 秒と延ばす（jitter 有効時はこれを上限にランダム）
retry_backoff_max=600, # 間隔の上限は600秒
retry_jitter=True, # 間隔にばらつきを入れて集中を避ける
max_retries=5,
)
def notify_webhook(self, url: str, payload: dict) -> int:
logger.info("通知送信 %s（%d 回目）", url, self.request.retries + 1)
resp = requests.post(url, json=payload, timeout=10)
resp.raise_for_status()
return resp.status_code
@app.task
def cleanup_old_files(directory: str, days: int = 7) -> int:
limit = time.time() - days * 86400
removed = 0
for path in Path(directory).glob("*.tmp"):
if path.stat().st_mtime < limit:
path.unlink()
removed += 1
return removedワーカーを起動します。-A には Celery アプリがあるモジュールを指定します。
celery -A proj.celery_app worker --loglevel=INFO起動ログの [tasks] 欄に proj.tasks.notify_webhook と proj.tasks.cleanup_old_files が表示されていれば、タスクが登録されています。別の端末からタスクを投入してみます。
from proj.tasks import notify_webhook
result = notify_webhook.delay("https://example.com/hook", {"post_id": 123})
print(result.id) # タスクID
print(result.get(timeout=30)) # 完了まで待って戻り値を受け取る（確認用）
# 60秒後に実行したい場合
notify_webhook.apply_async(
args=["https://example.com/hook", {"post_id": 124}],
countdown=60,
)result.get() は動作確認には便利ですが、Web アプリの中で呼ぶと結局そこで待つことになり、非同期にした意味がなくなります。また、タスクの中から別タスクの get() を呼ぶとワーカー同士が待ち合って止まるおそれがあり、Celery は既定でこれをエラーにします。
タスクを書くときの3つの原則
引数は小さく、JSON にできる値だけにする。 データベースのオブジェクトや大きなファイルの中身を渡すのではなく、ID やファイルパスを渡し、タスクの中で読み直します。キューに入ってから実行されるまでの間にデータが変わる可能性があるためです。
同じタスクが2回実行されても困らないようにする（冪等性）。 task_acks_late=True にすると、ワーカーが途中で落ちたタスクは再配信されます。取りこぼしは減りますが、同じ処理が重複して走ることがあります。「送信済みフラグを確認してから送る」など、重複しても結果が変わらない作りにしておきます。
時間制限を決めておく。 ソフト制限に達するとタスク内で SoftTimeLimitExceeded 例外が発生するので、後片付けをしてから終われます。ハード制限はワーカーがプロセスごと止めるため、後片付けはできません。
Redis をブローカーにする場合、visibility_timeout（既定は1時間）より長い countdown や eta を指定すると、タスクが再配信されて重複実行されることがあります。長い遅延実行が必要なら broker_transport_options でこの値を調整するか、Beat による定期実行で代替します。
Celery Beat で定期実行する
cron の代わりに Celery 側で定期実行を管理したい場合は Beat を使います。設定は beat_schedule に書きます。
# proj/celery_app.py の末尾に追加
from celery.schedules import crontab
app.conf.beat_schedule = {
"cleanup-tmp-every-night": {
"task": "proj.tasks.cleanup_old_files",
"schedule": crontab(hour=3, minute=0),
"args": ("/var/tmp/proj", 7),
},
}celery -A proj.celery_app beat --loglevel=INFOBeat は時刻になったらタスクをキューに入れるだけで、実行するのはワーカーです。Beat を複数台で起動すると同じタスクが重複して投入されるため、1つの環境で1プロセスだけ動かします。timezone を設定していれば、crontab の時刻はその時間帯で解釈されます。
動作確認と監視
ワーカーの状態はコマンドで確認できます。
celery -A proj.celery_app status # 応答しているワーカー一覧
celery -A proj.celery_app inspect active # 実行中のタスク
celery -A proj.celery_app inspect registered # 登録済みタスク名画面で確認したい場合は、監視ツールの Flower（pip install flower）が使えます。タスクの引数や結果が見えてしまうため、インターネットに直接公開せず、認証を付けるか SSH トンネル経由で見るようにしてください。
よくある症状と確認点は次のとおりです。
- **
Received unregistered taskと出る**：ワーカーがタスクのモジュールを読み込んでいません。includeの指定と、起動ログの[tasks]欄を確認します。コードを変更したらワーカーの再起動が必要です。 - タスクが PENDING のまま進まない：ワーカーが起動していないか、投入側とワーカーで接続先のブローカーが違う可能性があります。両方の
broker設定とcelery statusを確認します。なお、存在しないタスクIDも PENDING と表示されます。 - ブローカーに接続できない：
redis-cli pingで Redis 自体が応答するか、ファイアウォールやパスワード設定を確認します。
systemd で常駐させる
本番ではワーカーをサービスとして登録し、サーバー再起動後も自動で立ち上がるようにします。以下は例なので、パスやユーザー名、Redis のサービス名（ディストリビューションによって redis-server や redis など）は環境に合わせて変えてください。
# /etc/systemd/system/celery-worker.service
[Unit]
Description=Celery worker for proj
After=network.target redis-server.service
[Service]
Type=simple
User=celery
Group=celery
WorkingDirectory=/opt/proj
EnvironmentFile=/etc/proj/celery.env
ExecStart=/opt/proj/venv/bin/celery -A proj.celery_app worker --loglevel=INFO --concurrency=4
Restart=always
TimeoutStopSec=300
[Install]
WantedBy=multi-user.targetsudo systemctl daemon-reload
sudo systemctl enable --now celery-worker
journalctl -u celery-worker -f # ログを追うワーカーは停止の合図を受けると、実行中のタスクが終わるのを待ってから終了します。TimeoutStopSec はタスクの最大実行時間より長めにしておくと、デプロイ時の再起動で処理が途中で切られにくくなります。Beat を使う場合は、同じ要領で別のサービスとして登録します。ワーカーは root ではなく専用ユーザーで動かし、環境変数ファイルの権限も絞っておきます。
まとめ
- Celery はプロデューサー・ブローカー・ワーカー・結果バックエンドの組み合わせで動く。動かないときは各部品を順に確認する
- シリアライザーは JSON に限定し、接続情報は環境変数で渡す
- タスクの引数は ID などの小さな値にし、重複実行されても困らない作りにする
- リトライは
autoretry_forとretry_backoff、長時間化はtask_time_limitで制御する - 定期実行は Beat を1プロセスだけ動かし、本番のワーカーは systemd で常駐させる