Django集成Celery实现定时任务配置指南
发布时间：2026-10-13 |
4人浏览过
本文详细解析在Django项目中集成Celery实现定时任务的完整流程。涵盖celery.py配置、settings.py参数设置、__init__.py初始化以及task.py任务定义。通过具体代码示例，展示如何配置消息代理、结果后端及定时调度策略，帮助开发者快速搭建高效的异步任务系统。
创建Celery应用配置文件
首先，在项目同名目录下创建名为celery.py的文件。该文件用于初始化Celery应用实例并配置基础运行环境，包括设置Django的配置模块路径，确保Celery Worker能够正确读取项目配置。
from __future__ import absolute_import
import os
from celery import Celery
from datetime import timedelta
from kombu import Queue
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'OpsManage.settings')
from django.conf import settings
app = Celery('OpsManage')
# Using a string here means the worker will not have to
# pickle the object when using Windows.
# 配置celery
class Config:
BROKER_URL = 'amqp://guest:guest@localhost:5672//'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_RESULT_EXPIRES = 60 * 60
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = True
CELERY_ANNOTATIONS = {'*': {'rate_limit': '500/s'}}
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
app.config_from_object(Config)
# 到各个APP里自动发现tasks.py文件
app.autodiscover_tasks()
#crontab config
app.conf.update(
CELERYBEAT_SCHEDULE = {
# 每隔30s执行一次函数
'every-30-min-add': {
'task': 'apps.tasks.celery_assets.push_host_by_salt_tasks',
'schedule': timedelta(seconds=30)
# # 每天凌晨12点
# 'schedule': crontab(minute=0, hour=0)
},
},
)
# kombu : Celery 自带的用来收发消息的库, 提供了符合 Python 语言习惯的, 使用 AMQP 协议的高级接口
Queue('transient', routing_key='transient',delivery_mode=1)配置Django设置文件
接着，在项目的settings.py文件中进行Celery相关参数的配置。需要注册django_celery_beat和django_celery_results应用，以便利用Django后台管理定时任务调度及查看执行结果。
INSTALLED_APPS = [
......
'django_celery_beat',
'django_celery_results',
]初始化Celery应用
然后，在项目同名目录下的__init__.py文件中导入并声明Celery应用。这一步至关重要，它确保Django启动时Celery应用被自动加载，从而使得shared_task装饰器能够正确关联到该应用实例。
# coding:utf-8
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']
import pymysql
pymysql.install_as_MySQLdb()定义并注册定时任务
最后，在task.py文件中定义具体的任务函数，并使用@task装饰器将其标记为Celery任务。通过这种方式，可以将耗时操作或定时执行的逻辑封装在函数中，由Celery Worker异步执行。
from celery import task
# 定时任务
@task
def push_host_by_salt_tasks():
“”“balabala”“”
return '这里是定时任务'以上即为在Django中配置和使用Celery实现定时任务的核心步骤。通过合理配置消息代理、后端存储及任务调度，可以有效提升Web应用的响应速度和并发处理能力。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
