当前位置：首页 > news >正文

使用django_celery_beat在admin后台配置计划任务

news 2025/8/28 12:33:50

一、依赖包的安装

django中使用celery做异步任务和计划任务最头疼的点就是包之间版本兼容性问题，项目一启动花花报错，大概率都是版本问题。每次都会花很大时间在版本兼容性问题上。本例使用如下版本：

Django==3.2

celery==5.2.7

django-celery-beat==2.2.1

django-celery-results==2.4.0

django-redis==4.11.0

redis==3.5.3

如果是其他版本的django且有兼容性问题，推荐一个工具pipdeptree，这玩意可以清晰展示包间依赖关系和版本冲突，很方便。

二、主要配置

setting.py:

INSTALLED_APPS = ['django_celery_results','django_celery_beat',
]

tasks.py:

from itom_service.celery_app import app, logger@app.task()
def print_123(**kwargs):logger.info('开始打印')print(kwargs)

celery_app.py:

import osfrom celery.schedules import crontab
from celery.utils.log import get_logger
from django.conf import settingsfrom kombu import Queue, Exchange
from celery import Celerylogger = get_logger('celery')os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'itom_service.settings.settings')app = Celery('celery', broker=settings.CELERY_BROKER_URL, backend=settings.CELERY_RESULT_BACKEND)
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)app.conf.update(# enable CELERY_ALWAYS_EAGER=True can debug you tasks# CELERY_ALWAYS_EAGER=True,CELERY_TASK_RESULT_EXPIRES=None,  # 任务结果永不过期CELERY_TASK_SERIALIZER='json',CELERY_ACCEPT_CONTENT=['json'],CELERY_RESULT_SERIALIZER='json',CELERYD_CONCURRENCY=10,CELERYD_FORCE_EXECV=True,CELERYD_MAX_TASKS_PER_CHILD=100,CELERY_DEFAULT_QUEUE='default',CELERY_DEFAULT_ROUTING_KEY='default',CELERY_QUEUES=(Queue('default', Exchange('default'), routing_key='default'),Queue('demo_queue', Exchange('demo_queue'), routing_key='demo_queue'),),CELERY_TIMEZONE=settings.TIME_ZONE,# 关键配置：配合django_celery_beat使用，admin配置的定时任务可以被顺利执行CELERYBEAT_SCHEDULER='django_celery_beat.schedulers:DatabaseScheduler', # 这配置的计划任务会被自动同步到数据库，可以在admin后台进行编辑CELERYBEAT_SCHEDULE={# schedule1：execute tasks every five minutes(sync_host)'sync_host': {"task": "itom_service.apps.user.tasks.print_123","schedule": crontab(minute='*/1'),"args": (),"kwargs": {'key': '这是写死的定时任务'}},}
)