Django实现定时任务的两种方案：Celery与django-crontab详解
本文详细介绍在Django框架中实现定时任务的两种主流方案。首先讲解基于Celery和Redis的高级异步任务处理，涵盖环境配置、celery.py设置及后台管理；其次介绍轻量级的django-crontab方案，通过修改settings.py实现简单的Cron任务调度。内容包含具体代码示例、命令参数说明及注意事项，适用于不同复杂度的项目需求。
Django作为主流的后端Web开发框架，在实际应用中常需要执行固定频率或特定时间触发的代码逻辑。为了实现这一需求，开发者通常会引入Celery中间件，即Django-celery组件，以处理异步和定时任务。
本文所基于的运行环境如下：
- Python 3.6
- Django版本需小于1.8
- Celery版本为3.1
第一步：安装django-celery依赖包。
pip install django-celery
第二步：完成Celery配置及任务定义。
首先创建一个用于测试的Django项目环境。
django-admin.py createproject test django-admin.py startapp demo
项目创建完成后的目录结构如下所示：
- proj/ - manage.py - proj/ - __init__.py - celery.py - settings.py - urls.py - demo/ - migrations - __init__.py - admin.py - apps.py - models.py - tasks.py - tests.py - views.py
2.1 配置celery.py文件。
代码中需要替换的部分已在对应行后添加注释提示，其余部分保持默认配置即可。
在项目目录下创建test/test/celery.py文件，具体代码如下：
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') # “proj.settings”替换为你的项目信息：test.settings
app = Celery('proj') # 这里的proj替换为你的项目名称:test
# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
# should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')
# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
2.2 在项目的__init__.py中配置Celery应用。
打开test/test/__init__.py文件，并添加以下导入代码：
from __future__ import absolute_import, unicode_literals
# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app
__all__ = ('celery_app',)
2.3 在task.py文件中定义计划任务。
编辑test/demo/task.py文件，添加具体的定时任务逻辑，代码如下：
# Create your tasks here from __future__ import absolute_import, unicode_literals from celery import shared_task @shared_task def add(x, y): return x + y @shared_task def mul(x, y): return x * y @shared_task def xsum(numbers): return sum(numbers)
第三步：执行任务。
启动Django项目，命令为：python manage.py runserver。
3.1 在后台界面添加计划任务。
访问“http://localhost:8000/admin/”，进入Celery管理页面，选择“Periodic tasks”进行任务添加。选中已定义的任务，并设置具体的定时或周期时间。
3.2 启动Celery定时服务。
注意：Celery依赖Redis服务，因此在启动前需确保Redis服务已运行，命令为：`redis-server`。
# 以下两个命令在不同的shell窗口里执行，需要在django的目录下 python manager.py celery beat -l info #接收定时任务的命令 python manager.py celery worker -l info #执行定时任务的命令，此shell窗口会看到任务的输入信息
3.3 启动单次执行的Celery服务。
注意：同样依赖Redis服务，启动前需确保Redis服务已运行，命令为：`redis-server`。
python manager.py shell # 进到django的shell里 from demo.task import mul, xsum # 导入task任务 a = mul() b = xsum() # 执行a， b会输出信息 a(1,2) b(1)
补充方案：使用django-crontab实现Django定时任务。
安装django-crontab库。
pip install django-crontab
将django-crontab添加到settings.py的INSTALLED_APPS中，配置代码如下：
INSTALLED_APPS = ( 'django_crontab', ... )
在settings.py中配置django-crontab的具体命令规则：
CRONJOBS = [
('47 11 * * *', 'django.core.management.call_command', ['closepoll']，{},'>> /var/run.log'),
]
配置格式说明如下：
参数1：Cron定时表达式，例如“47 11 * * *”表示每天11时47分执行。 参数2：Python模块路径，若执行django-admin命令，则填写“django.core.management.call_command”。 参数3：方法的位置参数列表（默认值为[]），若执行django-admin命令，则填写所需命令，例如在polls中定义的closepoll。 参数4：方法的关键字参数字典（默认值为{}）。 参数5：日志存放位置（即重定向输出到文件，默认为空字符串）。
加载django-crontab任务。
任务加载过程较为简单，只需运行命令：python manage.py crontab add。
查看当前已激活的任务，使用命令：python manage.py crontab show。
删除已有任务，使用命令：python manage.py crontab remove。
若修改了任务配置，务必重新运行python manage.py crontab add以更新定时任务。
以上即为本文的全部内容，希望对读者有所帮助，欢迎继续支持本站。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
