Celeryの使い方｜Redis/Valkeyで非同期タスクを動かす

Celery は、重い処理を「タスク」として別プロセス（worker）に渡し、Web アプリの応答を待たせないための Python のライブラリです。 タスクはブローカー（Redis・Valkey・RabbitMQ などのキュー）に積まれ、worker が取り出して実行します。結果を受け取るときは結果バックエンド（result backend）に保存します。

最小の形は次の3つです。

# tasks.py
from celery import Celery

app = Celery("tasks",
             broker="redis://localhost:6379/0",    # タスクを積む場所
             backend="redis://localhost:6379/1")   # 結果を置く場所

@app.task
def add(x, y):
    return x + y

celery -A tasks worker -l INFO          # worker を起動

>>> from tasks import add
>>> r = add.delay(2, 3)     # キューに積むだけ。すぐ戻る
>>> r.get(timeout=10)       # 結果を待つ
5

確認環境：Celery 5.6.3・kombu 5.6.2・redis-py 8.1.0・Python 3.11（Ubuntu 24.04）。ブローカーと結果バックエンドには apt で入れた Valkey 7.2.13 を使い、worker は prefork・並列数 2 で動かしました。本文の出力・ログ・秒数は、この環境で得たものです。設定の既定値は Celery 公式ドキュメントで確かめました（確認日 2026年10月13日）。

Celery の仕組み：4つの役割

worker を止めた状態で add.delay() を5回呼ぶと、Valkey のリスト celery の長さが 5 になりました。 タスクは worker が起動するまでキューで待ちます。

$ valkey-cli llen celery
(integer) 5
$ celery -A tasks purge -f
Purged 5 messages from 1 known task queue.

Valkey をブローカーにする手順

Valkey は Redis 7.2 から分かれたオープンソースのキー・バリュー型データベースで、Redis と同じプロトコルで話します。 Celery 側は redis:// の URL のまま、Valkey に接続できました。

sudo apt install valkey-server       # Ubuntu 24.04 では 7.2.13 が入った
valkey-cli ping                      # PONG
pip install "celery[redis]"          # celery と redis-py が入る

worker を起動すると、接続先と登録されたタスクが表示されます。

 -------------- celery@vm v5.6.3 (recovery)
- ** ---------- .> transport:   redis://localhost:6417/0
- ** ---------- .> results:     redis://localhost:6417/1
- *** --- * --- .> concurrency: 2 (prefork)
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
[tasks]
  . tasks.add
  . tasks.flaky
  ...
[INFO/MainProcess] Connected to redis://localhost:6417/0
[INFO/MainProcess] celery@vm ready.

ブローカーの URL の形は、公式ドキュメントでは redis://:password@hostname:port/db_number です。省略すると localhost・6379・DB 0 になります。

root で起動すると SecurityWarning: You're running the worker with superuser privileges が出ました。本番では専用のユーザーで動かすか、--uid で実行ユーザーを指定します。

完成コード：tasks.py

この記事で動かしたタスクを1つにまとめたものです。

# tasks.py
import time
from celery import Celery
from celery.schedules import crontab

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
app.conf.update(
    timezone="Asia/Tokyo",
    task_acks_late=True,            # 実行が終わってから ack（取り出し完了）を返す
    worker_prefetch_multiplier=1,   # 1プロセスが先取りするのは1件
    result_expires=3600,            # 結果を1時間で消す
    task_track_started=True,        # 実行中を STARTED で見えるようにする
    beat_schedule={
        "heartbeat-every-10s": {"task": "tasks.heartbeat", "schedule": 10.0},
        "report-9am": {"task": "tasks.heartbeat", "schedule": crontab(hour=9, minute=0)},
    },
)

class TemporaryError(Exception):
    pass

@app.task
def add(x, y):
    return x + y

@app.task
def mul(x, y):
    return x * y

@app.task
def tsum(numbers):
    return sum(numbers)

@app.task
def slow(sec):
    time.sleep(sec)
    return f"slept {sec}s"

@app.task(bind=True, max_retries=3, default_retry_delay=2)
def flaky(self, fail_times):
    if self.request.retries < fail_times:
        raise self.retry(exc=TemporaryError(f"try {self.request.retries}"))
    return f"ok after {self.request.retries} retries"

@app.task(bind=True, autoretry_for=(TemporaryError,), retry_backoff=1,
          retry_backoff_max=8, retry_jitter=False, max_retries=4)
def auto(self):
    print(f"auto: retries={self.request.retries} at {time.strftime('%H:%M:%S')}")
    raise TemporaryError("always fails")

@app.task(soft_time_limit=2, time_limit=4)
def limited(sec):
    from celery.exceptions import SoftTimeLimitExceeded
    try:
        time.sleep(sec)
    except SoftTimeLimitExceeded:
        return "soft limit: cleaned up"
    return "done"

@app.task
def heartbeat():
    return time.strftime("%H:%M:%S")

celery -A tasks worker -l INFO -c 2                 # worker（並列数2）
celery -A tasks beat -l INFO                        # 定期実行（別のターミナル）

delay・apply_async・countdown と状態

r = add.delay(2, 3)
print(r.id, r.state)            # e823ecff-... PENDING
print(r.get(timeout=10), r.state)   # 5 SUCCESS

r2 = add.apply_async((10, 20), countdown=3)
r2.get(timeout=10)              # 30（呼んでから 3.0 秒後）

PENDING は「まだ知らない」という意味でもあります。 存在しない ID の結果を見ても PENDING になる仕様なので、「キューにある」とは限りません。task_track_started の既定は無効で、無効のままだと実行中も PENDING のままです。

結果は Valkey に JSON で入っていました。result_expires=3600 にしたので、TTL（残り秒数）が付いています。

$ valkey-cli -n 1 get celery-task-meta-766ca5af-...
{"status": "SUCCESS", "result": 9, "traceback": null, "children": [], "date_done": "2026-10-13T15:48:50.092488+09:00", ...}
$ valkey-cli -n 1 ttl celery-task-meta-766ca5af-...
(integer) 3508

apply_async のよく使う引数です（公式ドキュメント「Calling Tasks」より）。

retry：失敗したら再実行する

self.retry で再実行する

bind=True にすると、関数の1つ目の引数 self でタスク自身を受け取れます。 self.request.retries が今の再試行回数、self.retry() が再実行の予約です。

f = flaky.delay(2)      # 2回失敗してから成功する
f.get(timeout=30)       # 'ok after 2 retries'（4.5 秒）

worker のログでは、default_retry_delay=2 のとおり 2 秒ごとに再実行されていました。

15:48:24,427 Task tasks.flaky[0c75c411-...] retry: Retry in 2s: TemporaryError('try 0')
15:48:26,429 Task tasks.flaky[0c75c411-...] retry: Retry in 2s: TemporaryError('try 1')
15:48:28,431 Task tasks.flaky[0c75c411-...] succeeded in 0.002s: 'ok after 2 retries'

再試行のたびに received のログが出ます。 retry は「同じ ID のタスクをもう一度キューに積む」動きです。

5回失敗するように呼ぶと、max_retries=3 を超えたところで止まりました。 exc= を渡しているので、最後は渡した例外がそのまま上がり、状態は FAILURE です。

15:48:34,456 ERROR Task tasks.flaky[45fcd56c-...] raised unexpected: TemporaryError('try 3')

exc= を渡さずに self.retry() を呼んだ場合は、公式ドキュメントでは回数を超えると MaxRetriesExceededError になるとされています。

autoretry_for で自動的に再実行する

autoretry_for に例外を書くと、try/except を書かずに再実行されます。 retry_backoff を付けると、待ち時間が倍々に伸びます。

@app.task(bind=True, autoretry_for=(TemporaryError,), retry_backoff=1,
          retry_backoff_max=8, retry_jitter=False, max_retries=4)
def auto(self):
    raise TemporaryError("always fails")

worker のログの待ち時間は 1秒 → 2秒 → 4秒 → 8秒 で、5回目（retries=4）の失敗で FAILURE になりました。

15:48:34,461 auto: retries=0 at 15:48:34
15:48:34,467 Task tasks.auto[9afeba96-...] retry: Retry in 1s: TemporaryError('always fails')
15:48:35,467 auto: retries=1 at 15:48:35
15:48:37,476 auto: retries=2 at 15:48:37
15:48:37,480 Task tasks.auto[9afeba96-...] retry: Retry in 4s: TemporaryError('always fails')
15:48:41,483 auto: retries=3 at 15:48:41
15:48:41,487 Task tasks.auto[9afeba96-...] retry: Retry in 8s: TemporaryError('always fails')
15:48:49,498 auto: retries=4 at 15:48:49
15:48:49,501 Task tasks.auto[9afeba96-...] raised unexpected: TemporaryError('always fails')

（retries=1 のあとの Retry in 2s の行は省略しています。）

retry_jitter は既定で True です。 待ち時間を倍々の値どおりにしたいときは retry_jitter=False にします（上の例）。

時間制限（soft_time_limit・time_limit）

soft_time_limit を過ぎるとタスクの中で SoftTimeLimitExceeded が上がり、後片付けができます。 time_limit を過ぎると、worker のプロセスごと止められます。

limited.delay(5).get(timeout=10)   # 'soft limit: cleaned up'（2秒で打ち切り）

chain・group・chord：タスクをつなぐ

add.s(2, 3) は「シグネチャ」と呼ばれる、引数つきのタスクの呼び出し予約です。 これを組み合わせて流れを作ります。

from celery import chain, group, chord

chain(add.s(2, 3), mul.s(10), add.s(1))().get()        # 51  (2+3)*10+1
(add.s(1, 1) | mul.s(5)).delay().get()                  # 10
group(add.s(i, i) for i in range(5))().get()            # [0, 2, 4, 6, 8]
chord((mul.s(i, i) for i in range(1, 5)), tsum.s())().get()  # 30  1+4+9+16
chain(add.s(1, 2), add.si(100, 200))().get()            # 300

chord には結果バックエンドが要ります。 公式ドキュメントには「chord を使うには result_backend を有効にする」「chord の中のタスクは結果を無視してはいけない（ignore_result にしない）」とあります。

beat：定期実行

beat は、スケジュールに従ってタスクをキューに積むだけのプロセスです。 実行するのは worker なので、beat と worker の両方を動かします。

app.conf.beat_schedule = {
    "heartbeat-every-10s": {"task": "tasks.heartbeat", "schedule": 10.0},
    "report-9am": {"task": "tasks.heartbeat", "schedule": crontab(hour=9, minute=0)},
}
app.conf.timezone = "Asia/Tokyo"

celery -A tasks beat -l INFO -s ./celerybeat-schedule

24 秒動かすと、10 秒ごとに積まれ、worker が実行しました。

LocalTime -> 2026-10-13 15:49:45
    . scheduler -> celery.beat.PersistentScheduler
    . db -> ./celerybeat-schedule
[15:49:55,506] Scheduler: Sending due task heartbeat-every-10s (tasks.heartbeat)
[15:50:05,496] Scheduler: Sending due task heartbeat-every-10s (tasks.heartbeat)

beat は1つだけ動かします。 2つ動かすと、同じタスクが2回ずつ積まれます。-s で指定したファイル（ここでは celerybeat-schedule.db ができた）に、最後に実行した時刻が保存されます。

よく使う設定の一覧（既定値つき）

既定値は Celery 5.6 の公式ドキュメント「Configuration and defaults」で確かめたものです。

Redis/Valkey をブローカーにするときの「visibility timeout（見えなくなる時間）」の既定は1時間です。 公式ドキュメントには、countdown・eta・retry の待ち時間がこれを超えると、タスクが再び配られて重複して実行されることがあると書かれています。長い待ちを使うなら、broker_transport_options={"visibility_timeout": 秒数} を延ばします。

よくあるエラーと詰まりどころ

ログの全文です。

[ERROR/MainProcess] Received unregistered task of type 'tasks.not_exists'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

datetime.date を引数に渡したときは、エラーにならずに worker まで date 型のまま届きました。 kombu の JSON が日付を扱えるためです。届いた先で date + 1 をしたので、worker 側で TypeError: unsupported operand type(s) for +: 'datetime.date' and 'int' になりました。送る側でエラーにならなくても、受け取る型を決めておくと安全です。

タスクは2回以上実行されることがある前提で作ります。 公式ドキュメントでも、タスクは冪等（何回実行しても結果が同じ）にすることが勧められています。retry・acks_late・visibility timeout のどれでも、同じタスクがもう一度動くことがあります。「注文 ID で処理済みかを確かめてから書き込む」のような作りにします。非同期処理の設計の考え方は、関連記事「非同期処理の設計：同期との使い分け基準とジョブキューのリトライ設計」で扱っています。

運用で使うコマンド

worker を kill（SIGTERM）すると、worker: Warm shutdown (MainProcess) と出て、実行中のタスクを終えてから止まりました。

よくある質問

Q. Redis と Valkey のどちらでも同じコードで動きますか。 A. この記事では、redis:// の URL のまま Valkey 7.2.13 に接続し、ブローカーと結果バックエンドの両方で動きました。Redis 自体の使い方は、関連記事「Redisとは？使い方入門｜データ型・キャッシュ・よくある失敗を実行結果で解説」で扱っています。AWS のマネージドの Valkey は、関連記事「ElastiCacheの使い方｜Valkey/Redisの作り方と接続コード」で扱っています。

Q. FastAPI や Django から呼ぶには？ A. 呼ぶ側は add.delay() を呼んで、タスク ID をすぐ返すだけです。結果は別の API で AsyncResult(id).state を見る形にすると、応答を待たせません。FastAPI の基本は、関連記事「FastAPI入門｜uvで作るCRUD API完成コードと実行結果」で扱っています。

Q. asyncio とはどう違いますか。 A. asyncio は1つのプロセスの中で待ち時間を重ねる仕組みで、Celery は別プロセス・別サーバーの worker に仕事を渡す仕組みです。プロセスが落ちても、キューに残ったタスクは別の worker が処理できます。asyncio は関連記事「Python asyncioの使い方｜TaskGroupとtimeoutを実測で解説」で扱っています。

Q. テストで worker を立てずに動かせますか。 A. task_always_eager=True にすると、delay() がその場で実行されます。公式ドキュメントにある設定です。

まとめ

• Celery は producer → ブローカー → worker → 結果バックエンド の流れ。Valkey は redis:// のまま使えた
• delay は積むだけ。get() には result_backend が要る
• 再実行は self.retry()、自動なら autoretry_for + retry_backoff。待ち時間は 1→2→4→8 秒と伸びた
• chain は前の結果を次へ、group は並列、chord は並列のあとで集計（結果バックエンド必須）
• beat は 1つだけ。timezone を合わせ、worker も一緒に動かす
• タスクは 何回動いても同じ結果になるように作る

参考・出典

確認日はいずれも 2026年10月13日です。

• Celery ドキュメント「Tasks」（retry・autoretry_for・時間制限）：https://docs.celeryq.dev/en/stable/userguide/tasks.html
• Celery ドキュメント「Calling Tasks」：https://docs.celeryq.dev/en/stable/userguide/calling.html
• Celery ドキュメント「Canvas: Designing Work-flows」：https://docs.celeryq.dev/en/stable/userguide/canvas.html
• Celery ドキュメント「Periodic Tasks」：https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html
• Celery ドキュメント「Configuration and defaults」：https://docs.celeryq.dev/en/stable/userguide/configuration.html
• Celery ドキュメント「Using Redis」（visibility timeout）：https://docs.celeryq.dev/en/stable/getting-started/backends-and-brokers/redis.html
• celery（PyPI）：https://pypi.org/project/celery/
• Valkey：https://valkey.io/

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。