Python Celeryの使い方｜Redisで重い処理を非同期・並列実行する手順とmacOSの既知の不具合

Celery（セロリ）は、時間のかかる処理をWebアプリなどの本体から切り離し、別プロセスの「ワーカー」にバックグラウンドで実行させるPythonのタスクキューです。動かすには、Celery本体のほかにRedisやRabbitMQなどの「ブローカー」が必要です。

この記事では、CeleryとRedisの関係、最小構成での動かし方、並列実行・再試行・定期実行を、実際にワーカーを起動して確かめた結果つきで解説します。検証環境はmacOS・Python 3.12・Celery 5.6.3・Redis 8.10です（Celery 5.6.3は2026年10月15日時点のPyPI上の最新安定版）。macOSで最新版を使うと全タスクが失敗する既知の不具合があり、その回避方法も載せています。

Celeryとは：何をするライブラリか

たとえば「アップロードされた動画を変換する」「1万件にメールを送る」といった処理をWebのリクエストの中で実行すると、終わるまでユーザーを待たせ、タイムアウトの原因にもなります。Celeryを使うと、Web側は「この処理をやっておいて」とキューに登録するだけで即座に応答を返し、実際の処理は裏で動くワーカーが引き受けます。

Celeryの構成要素は次の3つです。

役割	中身	この記事での例
クライアント（登録する側）	タスクをキューに入れるプログラム。Webアプリのビューなど	client.py
ブローカー	タスクを預かってワーカーに渡す仲介役	Redis
ワーカー	タスクを受け取って実行するプロセス	celery -A tasks worker

タスクの戻り値を受け取りたい場合は、結果を保存する「結果バックエンド」も設定します。Redisはブローカーと結果バックエンドの両方に使えます。

CeleryとRedisの関係：ブローカーの選び方

「celery redis」で検索されるとおり、ブローカーにはRedisがよく使われます。公式ドキュメントのブローカー比較表（2026年10月15日確認）では、安定版（Stable）はRabbitMQ・Redis・Amazon SQSの3つで、ZookeeperやKafkaなどは実験的（Experimental）扱いです。

ブローカー	状態	監視	リモート操作
RabbitMQ	Stable	対応	対応
Redis	Stable	対応	対応
Amazon SQS	Stable	非対応	非対応
Zookeeper／Kafka	Experimental	非対応	非対応

Redisは導入が手軽で結果バックエンドも兼ねられるため、学習や小規模な用途の最初の選択肢に向いています。公式ドキュメントは、Redisは小さなメッセージの高速な受け渡しに向く一方、大きなメッセージは詰まりの原因になると説明しています。タスクの引数には大きなデータそのものではなく、ファイルのパスやIDを渡すのが基本です。

インストールと準備

pip install "celery[redis]"

Redisサーバーは別に用意します。公式ドキュメントでは、Dockerならdocker run -d -p 6379:6379 redisで起動する方法が紹介されています。macOSならHomebrew、Ubuntuならaptでもインストールできます。

なお、公式ドキュメントには「資金が限られているためWindowsはサポートしない」と明記されています（2026年10月15日確認）。Windowsで試す場合はWSL2やDockerの上で動かすのが確実です。

最小構成で動かす

1. タスクを定義する（tasks.py）

時間のかかる処理の例として、n未満の素数を数える関数をタスクにします。@app.taskを付けた関数がタスクになります。

import math
import time
from celery import Celery

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
app.conf.timezone = "Asia/Tokyo"


@app.task
def count_primes(n):
    """n未満の素数を数える（時間のかかる処理の例）"""
    start = time.perf_counter()
    count = 0
    for x in range(2, n):
        if all(x % d for d in range(2, math.isqrt(x) + 1)):
            count += 1
    return {"n": n, "primes": count, "seconds": round(time.perf_counter() - start, 2)}


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

2. ワーカーを起動する

tasks.pyと同じフォルダで、別のターミナルからワーカーを起動します。

celery -A tasks worker --loglevel=info

起動すると、接続先と登録されたタスクの一覧が表示されます（左側の飾り文字を省いた抜粋）。

.> transport:   redis://localhost:6379/0
.> results:     redis://localhost:6379/1
.> concurrency: 4 (prefork)
[tasks]
  . tasks.add
  . tasks.count_primes

3. タスクを登録して結果を受け取る（client.py）

import time
from tasks import count_primes

t0 = time.perf_counter()
result = count_primes.delay(2_000_000)          # キューに入れるだけ
print("登録まで:", round(time.perf_counter() - t0, 3), "秒")
print("タスクID:", result.id)
print("状態:", result.status)                    # まだ終わっていない

print("結果:", result.get(timeout=120))          # 終わるまで待って結果を受け取る
print("状態:", result.status)
print("合計:", round(time.perf_counter() - t0, 1), "秒")

登録まで: 0.041 秒
タスクID: 71dcdcc9-c772-4177-aa8d-4f06ec25f0d1
状態: PENDING
結果: {'n': 2000000, 'primes': 148933, 'seconds': 6.82}
状態: SUCCESS
合計: 6.9 秒

delay()は0.04秒ほどで戻り、計算そのもの（約6.8秒）はワーカー側で行われました。Webアプリなら、ここでタスクIDだけを画面に返し、あとで状態を問い合わせる作りにします。公式ドキュメントも、get()で完了を待つと非同期の呼び出しが同期になってしまうため、実際にはあまり使わないと説明しています。終わったかどうかだけ知りたいときはresult.ready()を使います。

macOSで全タスクが失敗する場合（2026年10月時点の既知の不具合）

最初にこの環境で試したとき、ワーカーは起動するのに、タスクがすべて次のエラーで失敗しました。

ValueError: not enough values to unpack (expected 3, got 0)

これはCeleryのGitHub Issue #10683で報告されている不具合です。Celeryが使う並列処理ライブラリbilliardの4.3系が、macOSでのプロセスの起動方法を変更したことで、Celery 5.6.3の標準の並列方式（prefork）では全タスクが失敗します。Linuxは影響を受けません。修正は次の5.7系に入っており、メンテナーは5.6系を使う場合はbilliardを4.2系に固定するよう案内しています（2026年10月15日確認。PyPIでは5.7.0b1がベータ版として公開中）。

次のどちらかで回避できました。

# 方法1：billiardを4.2系に固定する（prefork のまま使える）
pip install "billiard<4.3"

# 方法2：並列方式をスレッドに変える
celery -A tasks worker --loglevel=info -P threads

方法1で billiard 4.2.4 に戻すと、上の client.py は正常に動きました。方法2のスレッド方式（-P threads）でもタスクは成功しますが、CPUを使う計算はPythonの仕組み上スレッドでは並列になりにくいので、開発中の動作確認向けと考えてください。

複数のタスクを並列に実行する（group）

ワーカーは標準でCPUの数だけ子プロセスを立ち上げ（今回は--concurrency=4で4つに指定）、複数のタスクを同時に処理します。groupでまとめて登録すると、並列に実行されます。chainを使うと、前のタスクの戻り値を次のタスクに渡して順番に実行できます。

import time
from celery import group, chain
from tasks import count_primes, add

# 4つを順番に実行した場合と、groupで並列に投げた場合を比べる
t0 = time.perf_counter()
for n in [500_000] * 4:
    count_primes.delay(n).get(timeout=120)
print("順番に待つ:", round(time.perf_counter() - t0, 1), "秒")

t0 = time.perf_counter()
results = group(count_primes.s(n) for n in [500_000] * 4)().get(timeout=120)
print("groupで並列:", round(time.perf_counter() - t0, 1), "秒")
print([r["primes"] for r in results])

# chain：前のタスクの戻り値を次のタスクの第1引数に渡す
print(chain(add.s(2, 3), add.s(10), add.s(100))().get(timeout=10))

順番に待つ: 3.7 秒
groupで並列: 1.0 秒
[41538, 41538, 41538, 41538]
115

同じ計算を4回、1つずつ完了を待って実行すると3.7秒、groupで同時に投げると1.0秒でした。chainの結果115は、2＋3＝5、5＋10＝15、15＋100＝115と順に計算されたものです。.s()はタスクと引数をまとめた「シグネチャ」を作るメソッドです。

失敗したら自動で再試行する（autoretry_for）

外部APIへの通信のように、一時的に失敗しうる処理には再試行を設定します。autoretry_forに指定した例外が起きると、Celeryが自動でタスクをやり直します。

@app.task(bind=True, autoretry_for=(ConnectionError,),
          retry_kwargs={"max_retries": 3, "countdown": 1})
def fetch(self, url):
    if self.request.retries < 2:          # 2回目までは失敗させる
        raise ConnectionError("一時的な通信エラー")
    return f"{url} を取得（再試行 {self.request.retries} 回）"

from tasks import fetch
print(fetch.delay("https://example.com").get(timeout=30))

https://example.com を取得（再試行 2 回）

ワーカーのログには、1秒おきに2回再試行してから成功した様子が残りました（抜粋）。

retry: Retry in 1s: ConnectionError('一時的な通信エラー')
retry: Retry in 1s: ConnectionError('一時的な通信エラー')
succeeded in 0.0008787499973550439s: 'https://example.com を取得（再試行 2 回）'

このほか、タスクごとに実行頻度を制限するrate_limitなども公式ドキュメントのTasksで確認できます。

決まった間隔で実行する（Celery beat）

cronのような定期実行には、スケジューラーのcelery beatを使います。tasks.pyにスケジュールを書き足します。

app.conf.beat_schedule = {
    "add-every-5-seconds": {
        "task": "tasks.add",
        "schedule": 5.0,                  # 5秒ごと
        "args": (1, 2),
    },
}

celery -A tasks beat --loglevel=info

beatを起動すると、5秒ごとにタスクが登録され、ワーカーが実行しました。

[14:20:22] Scheduler: Sending due task add-every-5-seconds (tasks.add)
[14:20:27] Scheduler: Sending due task add-every-5-seconds (tasks.add)

公式ドキュメントによると、スケジュールの時刻は既定でUTCとして扱われるため、「毎朝9時」のような指定をするならapp.conf.timezone = "Asia/Tokyo"を設定します。また、beatは1つだけ起動する必要があり、複数起動すると同じタスクが重複して登録されると注意されています。

Celeryを使うべき場面・使わなくてよい場面

状況	おすすめ
Webアプリで重い処理を裏に回したい・複数台のサーバーで分担したい	Celery
失敗時の再試行、実行状況の確認、定期実行もまとめて管理したい	Celery
1台のPCで、1回きりのスクリプトを速くしたいだけ	concurrent.futuresやmultiprocessing（標準ライブラリ）
通信待ちの多い処理を1プロセスで大量にさばきたい	asyncio（標準ライブラリ）

Celeryはブローカーとワーカーを常に動かしておく必要があるため、仕組みとしては重めです。単発のスクリプトなら標準ライブラリで足りることが多く、Webアプリの非同期処理や定期実行を本格的に運用する段階でCeleryを検討するのがよいでしょう。Webフレームワークの選び方は「Pythonフレームワーク比較」で扱っています。

よくある質問

Q. CeleryにRedisは必須ですか？

必須ではありません。ブローカーはRabbitMQやAmazon SQSも使えます。ただしRedisは導入が手軽で結果バックエンドも兼ねられるため、最初の構成としてよく使われます。

Q. CeleryはWindowsで動きますか？

公式ドキュメントにWindowsはサポートしないと明記されています。Windowsで試す場合はWSL2やDockerの上で動かすのが確実です。

Q. macOSでタスクが「not enough values to unpack (expected 3, got 0)」で失敗します。

Celery 5.6.3とbilliard 4.3系の組み合わせで起きる既知の不具合です（GitHub Issue #10683）。pip install “billiard<4.3" で4.2系に戻すか、ワーカーを -P threads で起動すると回避できます。

Q. delay()とapply_async()の違いは？

delay()はapply_async()の短縮形で、引数をそのまま渡せますが実行オプションは指定できません。countdown（何秒後に実行するか）などを付けたいときはapply_async()を使います。なお add(2, 2) のように普通に呼ぶと、ワーカーではなく今のプロセスで実行されます。

あわせて読みたい

学習の順序はPython学習のロードマップ、標準で使える機能の一覧はPython標準ライブラリのモジュール一覧にまとめています。

Pythonの関数・メソッドの全体像はPython基本関数・メソッド完全一覧にまとめています。