Web 任务队列（Celery）

分布式任务队列 Celery 的完整使用指南，涵盖从入门到生产部署的核心知识点。

为什么需要任务队列

Web 请求需要快速响应，但发送邮件、处理图片、报表等操作耗时较长。任务队列将其异步化，请求立即返回，后台 Worker 慢慢处理。

同步：请求 -> 耗时操作(5s) -> 响应
异步：请求 -> 投递任务(即时) -> 响应，Worker 后台执行

典型场景：邮件发送(1-5s)、图片处理(2-10s)、数据报表(10-60s)、定时任务、第三方 API 调用。

架构概述

Producer (Web) ──> Broker (Redis/RabbitMQ) ──> Worker (执行任务)
                                                   │
                                                   ▼
                                            Backend (结果存储)

Producer (Web) ──> Broker (Redis/RabbitMQ) ──> Worker (执行任务)
                                                   │
                                                   ▼
                                            Backend (结果存储)

Producer：生产任务，由 Web 应用将任务发布到 Broker
Broker：消息中间件，存储和分发任务（Redis / RabbitMQ）
Worker：工作进程，从 Broker 拉取任务并执行
Backend：结果存储，保存任务状态和返回值（Redis / 数据库）
Beat：调度器，按计划自动发送定时任务

安装与配置

bash

pip install celery[redis]

pip install celery[redis]

python

# tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    timezone='Asia/Shanghai',
    enable_utc=True,
    task_track_started=True,
    task_soft_time_limit=300,
    task_time_limit=600,
    worker_concurrency=4,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    result_expires=3600 * 24,
)

# tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    timezone='Asia/Shanghai',
    enable_utc=True,
    task_track_started=True,
    task_soft_time_limit=300,
    task_time_limit=600,
    worker_concurrency=4,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    result_expires=3600 * 24,
)

配置文件方式

python

# celery_app.py
from celery import Celery
app = Celery('myapp')
app.config_from_object('celeryconfig')

# celeryconfig.py
broker_url = 'redis://localhost:6379/0'
result_backend = 'redis://localhost:6379/1'
task_serializer = 'json'
timezone = 'Asia/Shanghai'
task_track_started = True
task_soft_time_limit = 300
task_time_limit = 600
worker_concurrency = 4
task_acks_late = True
worker_prefetch_multiplier = 1
result_expires = 3600 * 24
imports = ('tasks',)

# celery_app.py
from celery import Celery
app = Celery('myapp')
app.config_from_object('celeryconfig')

# celeryconfig.py
broker_url = 'redis://localhost:6379/0'
result_backend = 'redis://localhost:6379/1'
task_serializer = 'json'
timezone = 'Asia/Shanghai'
task_track_started = True
task_soft_time_limit = 300
task_time_limit = 600
worker_concurrency = 4
task_acks_late = True
worker_prefetch_multiplier = 1
result_expires = 3600 * 24
imports = ('tasks',)

Django 集成

python

# myproject/celery.py
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# myapp/tasks.py
from celery import shared_task

@shared_task
def add(x, y):
    return x + y

# myproject/celery.py
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# myapp/tasks.py
from celery import shared_task

@shared_task
def add(x, y):
    return x + y

任务定义与调用

python

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')

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

@app.task
def send_welcome_email(email, username):
    import time; time.sleep(2)
    return f"邮件已发送至 {email}"

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')

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

@app.task
def send_welcome_email(email, username):
    import time; time.sleep(2)
    return f"邮件已发送至 {email}"

调用方式

python

# delay() -- 最简
result = add.delay(4, 5)
print(result.id)

# apply_async() -- 高级控制
add.apply_async(args=(4, 5), countdown=10)              # 延迟 10 秒
add.apply_async(args=(4, 5), queue='high_priority')     # 指定队列
add.apply_async(args=(4, 5), expires=60)                # 过期时间

# 签名 (signature)
add.s(4, 5).delay()

# delay() -- 最简
result = add.delay(4, 5)
print(result.id)

# apply_async() -- 高级控制
add.apply_async(args=(4, 5), countdown=10)              # 延迟 10 秒
add.apply_async(args=(4, 5), queue='high_priority')     # 指定队列
add.apply_async(args=(4, 5), expires=60)                # 过期时间

# 签名 (signature)
add.s(4, 5).delay()

Flask 集成

python

@app.route('/register', methods=['POST'])
def register():
    data = request.get_json()
    send_welcome_email.delay(data['email'], data['username'])
    return jsonify({"status": "ok"}), 201

@app.route('/register', methods=['POST'])
def register():
    data = request.get_json()
    send_welcome_email.delay(data['email'], data['username'])
    return jsonify({"status": "ok"}), 201

FastAPI 集成

python

@app.post("/register")
async def register(email: str, username: str):
    task = send_welcome_email.delay(email, username)
    return {"task_id": task.id}

@app.post("/register")
async def register(email: str, username: str):
    task = send_welcome_email.delay(email, username)
    return {"task_id": task.id}

任务状态与结果

状态机

PENDING -> STARTED -> SUCCESS / FAILURE / RETRY

PENDING -> STARTED -> SUCCESS / FAILURE / RETRY

获取结果

python

from celery.result import AsyncResult
from tasks import app

result = AsyncResult(task_id, app=app)
result.status       # 当前状态
result.ready()      # 是否已完成
result.successful() # 是否成功
result.failed()     # 是否失败
result.result       # 返回值或异常
result.traceback    # 失败时的 Traceback

# 阻塞等待（慎用）
value = result.get(timeout=10)

from celery.result import AsyncResult
from tasks import app

result = AsyncResult(task_id, app=app)
result.status       # 当前状态
result.ready()      # 是否已完成
result.successful() # 是否成功
result.failed()     # 是否失败
result.result       # 返回值或异常
result.traceback    # 失败时的 Traceback

# 阻塞等待（慎用）
value = result.get(timeout=10)

高级任务特性

绑定 (bind=True)

python

@app.task(bind=True)
def process_data(self, data):
    task_id = self.request.id
    retries = self.request.retries
    print(f"任务 {task_id} 第 {retries} 次重试")
    return data * 2

@app.task(bind=True)
def process_data(self, data):
    task_id = self.request.id
    retries = self.request.retries
    print(f"任务 {task_id} 第 {retries} 次重试")
    return data * 2

重试

python

@app.task(bind=True, max_retries=3, default_retry_delay=10, acks_late=True)
def send_email(self, to, subject, body):
    try:
        get_mail_client().send(to=to, subject=subject, body=body)
        return {"status": "sent"}
    except ConnectionError as exc:
        countdown = 2 ** self.request.retries * 10  # 指数退避
        raise self.retry(exc=exc, countdown=countdown)
    except InvalidEmailError:
        return {"status": "failed", "reason": "invalid_email"}

@app.task(bind=True, max_retries=3, default_retry_delay=10, acks_late=True)
def send_email(self, to, subject, body):
    try:
        get_mail_client().send(to=to, subject=subject, body=body)
        return {"status": "sent"}
    except ConnectionError as exc:
        countdown = 2 ** self.request.retries * 10  # 指数退避
        raise self.retry(exc=exc, countdown=countdown)
    except InvalidEmailError:
        return {"status": "failed", "reason": "invalid_email"}

钩子

python

# 在任务类中定义
def on_success(self, retval, task_id, args, kwargs):
    notify_admin(f"任务 {task_id} 成功")

def on_failure(self, exc, task_id, args, kwargs, einfo):
    notify_admin(f"任务 {task_id} 失败: {exc}")

def on_retry(self, exc, task_id, args, kwargs, einfo):
    print(f"任务 {task_id} 第 {self.request.retries} 次重试")

# 在任务类中定义
def on_success(self, retval, task_id, args, kwargs):
    notify_admin(f"任务 {task_id} 成功")

def on_failure(self, exc, task_id, args, kwargs, einfo):
    notify_admin(f"任务 {task_id} 失败: {exc}")

def on_retry(self, exc, task_id, args, kwargs, einfo):
    print(f"任务 {task_id} 第 {self.request.retries} 次重试")

超时

python

@app.task(soft_time_limit=60, time_limit=90)
def api_call():
    return requests.get('https://api.example.com/data', timeout=30).json()

@app.task(soft_time_limit=60, time_limit=90)
def api_call():
    return requests.get('https://api.example.com/data', timeout=30).json()

定时任务（Beat）

python

from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-every-30-seconds': {
        'task': 'tasks.cleanup_temp_files',
        'schedule': 30.0,
    },
    'daily-report': {
        'task': 'tasks.generate_daily_report',
        'schedule': crontab(hour=3, minute=0),
    },
    'weekly-report': {
        'task': 'tasks.send_weekly_report',
        'schedule': crontab(hour=9, minute=0, day_of_week=1),
    },
}

from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-every-30-seconds': {
        'task': 'tasks.cleanup_temp_files',
        'schedule': 30.0,
    },
    'daily-report': {
        'task': 'tasks.generate_daily_report',
        'schedule': crontab(hour=3, minute=0),
    },
    'weekly-report': {
        'task': 'tasks.send_weekly_report',
        'schedule': crontab(hour=9, minute=0, day_of_week=1),
    },
}

crontab 常用表达式

python

crontab()                              # 每分钟
crontab(minute=0)                      # 每小时整点
crontab(hour=0, minute=0)             # 每天凌晨
crontab(minute='*/15')                # 每 15 分钟
crontab(hour='8,12,18', minute=0)     # 每天 8/12/18 点
crontab(hour=9, minute=0, day_of_week='1-5')  # 工作日 9 点

crontab()                              # 每分钟
crontab(minute=0)                      # 每小时整点
crontab(hour=0, minute=0)             # 每天凌晨
crontab(minute='*/15')                # 每 15 分钟
crontab(hour='8,12,18', minute=0)     # 每天 8/12/18 点
crontab(hour=9, minute=0, day_of_week='1-5')  # 工作日 9 点

启动

bash

celery -A tasks beat --loglevel=info
celery -A tasks worker --beat --loglevel=info  # 与 Worker 一起

celery -A tasks beat --loglevel=info
celery -A tasks worker --beat --loglevel=info  # 与 Worker 一起

任务路由与工作流

路由

python

app.conf.task_routes = {
    'tasks.send_email':    {'queue': 'email'},
    'tasks.process_image': {'queue': 'image'},
    'tasks.*':             {'queue': 'default'},
}

app.conf.task_routes = {
    'tasks.send_email':    {'queue': 'email'},
    'tasks.process_image': {'queue': 'image'},
    'tasks.*':             {'queue': 'default'},
}

bash

celery -A tasks worker -Q email --concurrency=2          # 指定队列
celery -A tasks worker -Q email,default --concurrency=4  # 多队列

celery -A tasks worker -Q email --concurrency=2          # 指定队列
celery -A tasks worker -Q email,default --concurrency=4  # 多队列

工作流

python

from celery import chain, group, chord

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

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

# chain -- 链式：(1+2)*3=9
result = (add.s(1, 2) | multiply.s(3)).delay()
result.get()  # 9

# group -- 并行
result = group(multiply.s(n, 2) for n in range(1, 6)).delay()
result.get()  # [2, 4, 6, 8, 10]

# chord -- 并行 + 回调
@app.task
def summarize(results):
    return {"total": sum(results), "average": sum(results)/len(results)}

result = chord(multiply.s(n, n) for n in range(1, 11))(summarize.s())
result.get()  # {"total": 385, "average": 38.5}

from celery import chain, group, chord

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

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

# chain -- 链式：(1+2)*3=9
result = (add.s(1, 2) | multiply.s(3)).delay()
result.get()  # 9

# group -- 并行
result = group(multiply.s(n, 2) for n in range(1, 6)).delay()
result.get()  # [2, 4, 6, 8, 10]

# chord -- 并行 + 回调
@app.task
def summarize(results):
    return {"total": sum(results), "average": sum(results)/len(results)}

result = chord(multiply.s(n, n) for n in range(1, 11))(summarize.s())
result.get()  # {"total": 385, "average": 38.5}

监控与管理

Flower

bash

pip install flower
celery -A tasks flower --port=5555
celery -A tasks flower --basic-auth=admin:password123  # 启用认证

pip install flower
celery -A tasks flower --port=5555
celery -A tasks flower --basic-auth=admin:password123  # 启用认证

提供：任务实时列表、Worker 管理、队列深度、耗时/成功率图表。

日志

python

from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)

@app.task
def task_with_logging():
    logger.info("开始执行")
    logger.error("出错了")

from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)

@app.task
def task_with_logging():
    logger.info("开始执行")
    logger.error("出错了")

健康检查

python

def check_celery_health():
    try:
        conn = app.connection()
        conn.ensure_connection(max_retries=1)
        conn.release()
        stats = app.control.inspect().stats()
        return {"status": "healthy", "workers": list(stats.keys()) if stats else []}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

def check_celery_health():
    try:
        conn = app.connection()
        conn.ensure_connection(max_retries=1)
        conn.release()
        stats = app.control.inspect().stats()
        return {"status": "healthy", "workers": list(stats.keys()) if stats else []}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

最佳实践

1. 幂等：多次执行与一次执行结果相同。

python

@app.task
def process_order(order_id):
    if redis_client.sismember("processed_orders", order_id):
        return {"status": "already_done"}
    do_process(order_id)
    redis_client.sadd("processed_orders", order_id)

@app.task
def process_order(order_id):
    if redis_client.sismember("processed_orders", order_id):
        return {"status": "already_done"}
    do_process(order_id)
    redis_client.sadd("processed_orders", order_id)

2. 参数可序列化：传递 ID，在任务内部获取对象。

python

# 错误
@app.task
def process_db(conn): pass
# 正确
@app.task
def process_user(user_id):
    db = get_db(); user = db.query(User).get(user_id)

# 错误
@app.task
def process_db(conn): pass
# 正确
@app.task
def process_user(user_id):
    db = get_db(); user = db.query(User).get(user_id)

3. ACK 模式：task_acks_late=True 让 Worker 崩溃时任务重新投递。

4. 并发控制：

bash

celery -A tasks worker --concurrency=4                  # CPU 密集型
celery -A tasks worker --concurrency=20 --pool=threads  # I/O 密集型
celery -A tasks worker --prefetch-multiplier=1          # 预取 1 个任务
celery -A tasks worker --max-tasks-per-child=1000       # 防内存泄漏

celery -A tasks worker --concurrency=4                  # CPU 密集型
celery -A tasks worker --concurrency=20 --pool=threads  # I/O 密集型
celery -A tasks worker --prefetch-multiplier=1          # 预取 1 个任务
celery -A tasks worker --max-tasks-per-child=1000       # 防内存泄漏

5. 限流：

python

@app.task(rate_limit='10/m')
def rate_limited_api_call(self, url):
    return requests.get(url).json()

@app.task(rate_limit='10/m')
def rate_limited_api_call(self, url):
    return requests.get(url).json()

6. 忽略不需要的结果：@app.task(ignore_result=True)。

7. 优雅关闭：kill -TERM <pid> 或 celery -A tasks control shutdown。

Docker 部署

Dockerfile

dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

celery[redis]>=5.4,<6.0
flower>=2.0,<3.0
redis>=5.0,<6.0

celery[redis]>=5.4,<6.0
flower>=2.0,<3.0
redis>=5.0,<6.0

docker-compose.yml

yaml

version: "3.9"

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s; timeout: 5s; retries: 5
    restart: unless-stopped

  worker:
    build: .
    command: celery -A tasks worker --loglevel=info --concurrency=4
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
      - CELERY_RESULT_BACKEND=redis://redis:6379/1
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped

  beat:
    build: .
    command: celery -A tasks beat --loglevel=info
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped

  flower:
    build: .
    command: celery -A tasks flower --port=5555 --basic-auth=admin:changeme
    ports:
      - "5555:5555"
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped

volumes:
  redis-data:

version: "3.9"

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s; timeout: 5s; retries: 5
    restart: unless-stopped

  worker:
    build: .
    command: celery -A tasks worker --loglevel=info --concurrency=4
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
      - CELERY_RESULT_BACKEND=redis://redis:6379/1
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped

  beat:
    build: .
    command: celery -A tasks beat --loglevel=info
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped

  flower:
    build: .
    command: celery -A tasks flower --port=5555 --basic-auth=admin:changeme
    ports:
      - "5555:5555"
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped

volumes:
  redis-data:

启动命令

bash

docker-compose up -d
docker-compose up -d --scale worker=5 worker  # 扩缩
docker-compose logs -f worker
docker-compose down

docker-compose up -d
docker-compose up -d --scale worker=5 worker  # 扩缩
docker-compose logs -f worker
docker-compose down

常见问题

问题	排查方法
任务未执行	`celery -A tasks status` 检查 Worker
Worker 断连	设置 `broker_connection_retry_on_startup=True`
任务重复	确保幂等，使用去重
内存泄漏	设置 `--max-tasks-per-child=1000`
任务积压	增加 Worker 或检查队列路由

总结

三要素：Broker（消息存储）、Worker（执行）、Backend（结果存储）
任务：@app.task，支持绑定、重试、超时
调用：delay() / apply_async() / signature
工作流：chain 链式、group 并行、chord 并行+回调
定时任务：Beat + crontab
生产：合理并发、ACK 模式、Flower 监控
部署：Docker 多容器编排，水平扩展

Web 任务队列（Celery） ​

为什么需要任务队列 ​

架构概述 ​

安装与配置 ​

配置文件方式 ​

Django 集成 ​

任务定义与调用 ​

调用方式 ​

Flask 集成 ​

FastAPI 集成 ​

任务状态与结果 ​

状态机 ​

获取结果 ​

高级任务特性 ​

绑定 (bind=True) ​

重试 ​

钩子 ​

超时 ​

定时任务（Beat） ​

crontab 常用表达式 ​

启动 ​

任务路由与工作流 ​

路由 ​

工作流 ​

监控与管理 ​

Flower ​

日志 ​

健康检查 ​

最佳实践 ​

Docker 部署 ​

Dockerfile ​

docker-compose.yml ​

启动命令 ​

常见问题 ​

总结 ​

Web 任务队列（Celery）

为什么需要任务队列

架构概述

安装与配置

配置文件方式

Django 集成

任务定义与调用

调用方式

Flask 集成

FastAPI 集成

任务状态与结果

状态机

获取结果

高级任务特性

绑定 (bind=True)

重试

钩子

超时

定时任务（Beat）

crontab 常用表达式

启动

任务路由与工作流

路由

工作流

监控与管理

Flower

日志

健康检查

最佳实践

Docker 部署

Dockerfile

docker-compose.yml

启动命令

常见问题

总结