Django后台任务调度怎么实现

wen python案例 2026-07-12 1

本文目录导读：

Django后台任务调度怎么实现

使用 Celery（最推荐，功能最强大）
使用 Django-Q（轻量级替代方案）
使用 APScheduler（简单场景）
使用 Huey（轻量级 Redis 队列）
选择建议
生产环境注意事项

在 Django 中实现后台任务调度，有几种常见且成熟的方法，选择哪种取决于任务的复杂度、需要的持久化能力以及是否依赖外部服务。

使用 Celery（最推荐，功能最强大）

Celery 是 Python 最流行的分布式任务队列，与 Django 集成非常完善。

安装配置

pip install celery[redis]  # 推荐使用 Redis 作为 Broker
# 或
pip install celery[rabbitmq]

基本使用

项目目录结构：

myproject/
├── myproject/
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   └── ...
├── myapp/
│   ├── tasks.py
│   └── ...

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()

settings.py 添加配置：

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_BEAT_SCHEDULE = {
    'send-report-every-30-minutes': {
        'task': 'myapp.tasks.send_email_report',
        'schedule': crontab(minute='*/30'),
    },
}

tasks.py 定义任务：

from celery import shared_task
from django.core.mail import send_mail
@shared_task
def send_email_report():
    send_mail(
        '每日报告',
        '这是你的每日报告内容。',
        'from@example.com',
        ['to@example.com'],
    )
    return "任务完成"
@shared_task
def add(x, y):
    return x + y

启动 Worker 和 Beat

# 启动 Worker（处理任务）
celery -A myproject worker -l info
# 启动 Beat（调度定时任务）
celery -A myproject beat -l info
# 同时启动 Worker 和 Beat（开发环境）
celery -A myproject worker -B -l info

调用任务

# 异步调用
add.delay(4, 4)
# 延迟调用
from datetime import timedelta
add.apply_async((4, 4), countdown=10)  # 10秒后执行
# 定时调用
from celery import current_app
current_app.send_task('myapp.tasks.add', args=[4, 4],
                     eta=datetime.now() + timedelta(hours=1))

使用 Django-Q（轻量级替代方案）

Django-Q 是一个原生 Django 任务队列，配置简单，不需要 Celery 那样的基础设施。

安装配置

pip install django-q

settings.py 配置：

INSTALLED_APPS = [
    # ...
    'django_q',
]
Q_CLUSTER = {
    'name': 'myproject',
    'workers': 4,           # 并行工作进程数
    'recycle': 500,         # 每个 worker 处理任务数后重启
    'timeout': 60,          # 单个任务超时时间（秒）
    'compress': True,
    'save_limit': 250,
    'queue_limit': 500,
    'cpu_affinity': 1,
    'label': 'Django Q',
    'redis': {
        'host': '127.0.0.1',
        'port': 6379,
        'db': 0,
    }
}
# 定时任务配置
Q_CLUSTER['cron'] = [
    # 每天凌晨3点执行
    ('0 3 * * *', 'myapp.tasks.daily_cleanup'),
]

定义任务

from django_q.models import Schedule
from django_q.tasks import async_task, schedule
def send_newsletter(user_id):
    # 你的任务逻辑
    return f"Newsletter sent to user {user_id}"
# 调用方式
async_task('myapp.tasks.send_newsletter', user_id=123)
# 定时执行
from datetime import timedelta, datetime
schedule(
    'myapp.tasks.send_newsletter',
    123,
    schedule_type=Schedule.CRON,
    cron_expression='0 0 * * *',  # 每天午夜
    next_run=datetime.now()
)

启动 Worker

python manage.py qcluster

使用 APScheduler（简单场景）

如果不想依赖外部服务，APScheduler 可以直接在 Django 进程中运行。

安装配置

pip install apscheduler

scheduler.py 设置：

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore
from django.conf import settings
def start_scheduler():
    scheduler = BackgroundScheduler()
    scheduler.add_jobstore(DjangoJobStore(), 'default')
    # 添加任务
    scheduler.add_job(
        my_task_function,
        trigger='interval',
        hours=1,
        id='task_id',
        replace_existing=True
    )
    scheduler.start()
    return scheduler

集成到 Django

apps.py 中启动：

from django.apps import AppConfig
class MyAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'
    def ready(self):
        if not settings.DEBUG:  # 生产环境才启动
            from .scheduler import start_scheduler
            start_scheduler()

使用 Huey（轻量级 Redis 队列）

Huey 是一个轻量级的任务队列,特别适合小项目。

pip install huey

配置：

from huey import RedisHuey
from huey.contrib.djhuey import task, periodic_task
from huey.crontab import crontab
huey = RedisHuey('myapp', host='localhost')
@periodic_task(crontab(minute='0', hour='3'))  # 每天凌晨3点
def daily_backup():
    # 备份逻辑
    pass
@task()
def send_email_task(to, subject, body):
    # 发送邮件
    pass

选择建议

方案	适用场景	优点	缺点
Celery	大型项目、复杂任务	功能完善、可靠性高、社区活跃	配置复杂、依赖较多
Django-Q	中型项目	配置简单、原生集成	功能较少、社区较小
APScheduler	简单定时任务	不依赖外部服务、轻量	不持久化、进程内运行
Huey	小项目快速开发	极简配置、性能好	只支持 Redis

生产环境注意事项

任务重试：Celery 支持自动重试

@shared_task(bind=True, max_retries=3)
def my_task(self):
    try:
        # 业务逻辑
        pass
    except Exception as e:
        raise self.retry(exc=e, countdown=60)

结果持久化：配置 result_backend 存储任务结果
监控告警：使用 Flower（Celery）或自定义监控
任务优先级：Celery 支持设置队列优先级

限流：防止任务积压

CELERY_RATE_LIMITS = {'myapp.tasks.send_report': '1/m'}  # 每分钟1次

对于大多数 Django 项目，我推荐从 Celery 开始，它是最成熟、最稳定的选择，如果项目较小或想快速上手，Django-Q 是个不错的轻量替代方案。