豪翔天下

Change My World by Program

0%

django-celery-beat定时任务配置

安装

1
pip install django-celery-beat

配置

  1. settings.py中添加如下配置
1
2
3
4
5
6
7
8
9
10
INSTALLED_APPS = (
...,
'django_celery_beat',
)

CELERY_BROKER_URL = 'sqla+sqlite:///' + os.path.join(BASE_DIR, 'db.sqlite3')
CELERY_IMPORTS = ['schedules'] # 指定需要在哪些目录扫描定时任务
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
CELERY_RESULT_BACKEND = 'django-db' # 这是下面的django-celery-results保存结果使用的
  1. 使用命令python manage.py migrate创建数据库迁移,会生成这几张表:
1
2
3
django_celery_beat.models.PeriodicTask # 周期性任务
django_celery_beat.models.IntervalSchedule # 间隔性任务
django_celery_beat.models.CrontabSchedule # 定时任务
  1. 创建启动文件,在manage.py所在目录创建一个启动文件celery.py:
1
2
3
4
5
6
7
8
9
10
import os

from celery import Celery

from life_console import settings

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "项目名.settings")
celery_app = Celery("项目名", broker=settings.CELERY_BROKER_URL)
celery_app.config_from_object("django.conf:settings", namespace="CELERY")
celery_app.autodiscover_tasks()

创建定时任务

一个定时任务可以是一个简单的函数,例如(得在CELERY_IMPORTS下才能在后台创建),例如

1
2
3
4
5
6
from 项目名.celery import celery_app


@celery_app.task
def my_cron_reminder():
print(my_cron_reminder)

创建定时执行的任务:

  1. 在admin后台创建PeriodicTask,选择执行周期即可

  2. 或者手动插入至PeriodicTask:

    1
    2
    3
    schedule, created = IntervalSchedule.objects.get_or_create(every=10, period=IntervalSchedule.SECONDS)
    PeriodicTask.objects.create(interval=schedule, name='Custom task', task='schedules.my_corn_minder')
    PeriodicTask.objects.create(interval=schedule, name='Custom task', task='schedules.my_corn_minder', args=json.dumps(['arg1', 'arg2']), kwargs=json.dumps({'abc': 12})) # 提供参数

启动任务

需要使用下面的命令来启动worker,这样可以动态更改项目执行计划后自动生效

1
./venv/bin/celery -A 项目名 worker --beat --scheduler django -l debug --logfile=/var/log/celery.log --detach

使用Django-Celery-Result保存定时任务执行结果

安装: pip install django-celery-resutls && python manage.py migrate,这会生成一张taskresult表,上面定时任务的结果会自动保存在表中

坚持原创技术分享,谢谢支持

欢迎关注我的其它发布渠道