关于Django使用django-celery-beat动态添加定时任务的方法

⽅法

版本信息

# 插件安装 Django==2.2.2

django-celery-beat==2.1.0django-redis==4.8.0mysqlclient==2.0.0django-mysql==3.2.0redis==3.2.1

uWSGI==2.0.17.1

django-redis-cache==2.1.0

安装与配置

1. 安装上⾯的对应的celery版本2. 配置settings.py

# django时区配置

TIME_ZONE = 'Asia/Shanghai'

# 如果USE_TZ设置为True时，Django会使⽤系统默认设置的时区，此时的TIME_ZONE不管有没有设置都不起作⽤# 如果USE_TZ 设置为False,TIME_ZONE = 'Asia/Shanghai', 则使⽤上海的UTC时间。USE_TZ = FalseINSTALLED_APPS = ( ...,

'django_celery_beat',)

# celery beat配置

# CELERY_ENABLE_UTC = FalseCELERY_TIMEZONE = TIME_ZONE

DJANGO_CELERY_BEAT_TZ_AWARE = False

CELERY_BEAT_SCHEDULER = 'django-celery-beat.schedulers.DatabaseScheduler'# celery 的启动⼯作数量设置

CELERY_WORKER_CONCURRENCY = 10

# 任务预取功能，会尽量多拿 n 个，以保证获取的通讯成本可以压缩。CELERYD_PREFETCH_MULTIPLIER = 20# 有些情况下可以防⽌死锁

CELERYD_FORCE_EXECV = True

# celery 的 worker 执⾏多少个任务后进⾏重启操作

CELERY_WORKER_MAX_TASKS_PER_CHILD = 100

# 禁⽤所有速度限制，如果⽹络资源有限，不建议开⾜马⼒。CELERY_DISABLE_RATE_LIMITS = True# 设置代理⼈broker

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/2'# 指定 Backend

CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'

⽣成数据库

python manage.py migrate

# 迁移之后⽣成的表结构

django_celery_beat.models.PeriodicTask 此模型定义要运⾏的单个周期性任务。

django_celery_beat.models.IntervalSchedule 以特定间隔（例如，每5秒）运⾏的计划。

django_celery_beat.models.CrontabSchedule 与像在cron项领域的时间表 分钟⼩时⽇的⼀周 DAY_OF_MONTH month_of_yeardjango_celery_beat.models.PeriodicTasks 此模型仅⽤作索引以跟踪计划何时更改

在⼯作⽬录下配置celery.py

# -*- coding: utf-8 -*-# @File: celeryc.py

# @Content: celery定时任务配置

import os

from celery import Celery, platformsfrom celery.schedules import crontabfrom django.conf import settings

os.environ.setdefault(\"DJANGO_SETTINGS_MODULE\app = Celery(\"django_server\")

app.config_from_object(\"django.conf:settings\# 定时任务的存放位置

app.autodiscover_tasks([\"monitoring.tasks\

创建tasks任务

from celery import shared_task@shared_task

def alarm_monitor_task(**kwargs): print(\"定时任务\")

创建定时任务

from django_celery_beat.models import PeriodicTask, IntervalSchedule

-----周期性任务

# 创建10分钟的间隔 interval 对象

schedule, _ = IntervalSchedule.objects.update_or_create(every=10, period=IntervalSchedule.MINUTES)# 如果任务存在就更新，不存在则创建PeriodicTask.objects.update_or_create( defaults={

\"interval\": schedule, # 上⾯创建10分钟的间隔 interval 对象

\"task\": \"monitoring.tasks.alarm_monitor_task\指定需要周期性执⾏的任务 \"args\"=json.dumps(['arg1', 'arg2']),

\"kwargs\": json.dumps({\"a\": 1, \"b\": 2}, ensure_ascii=False) # 传⼊的参数 },

name=\"定时任务-task\)

# 周期性任务可选参数

IntervalSchedule.DAYS 固定间隔天数IntervalSchedule.HOURS 固定间隔⼩时数IntervalSchedule.MINUTES 固定间隔分钟数IntervalSchedule.SECONDS 固定间隔秒数

IntervalSchedule.MICROSECONDS 固定间隔微秒

----Crontab 周期性任务

from django_celery_beat.models import CrontabSchedule, PeriodicTask# 创建间隔30分钟执⾏的任务

crontab, _ = CrontabSchedule.objects.update_or_create( minute=\"*/30\ hour=\"*\

day_of_week=\"*\ day_of_month='*', month_of_year='*',

timezone=pytz.timezone(\"Asia/Shanghai\"),)

# 任务存在则更新，不存在创建

PeriodicTask.objects.update_or_create( name=task_name, defaults={

\"kwargs\": json.dumps(kwargs, ensure_ascii=False), \"task\": \"wechat.tasks.subscribe_task\ \"crontab\": crontab, },)

# 删除任务

task = PeriodicTask.objects.filter(name__startswith=sub_id)if task:

task.update(enabled=False) task.delete()

# 暂停当前任务

tasks = PeriodicTask.objects.filter(name__startswith=sub_id)if tasks:

tasks.update(enabled=True if status else False)

运⾏任务

# 启动任务 work

celery -A django_server worker -l INFO --logfile=/var/log/dec_server/worker.log

# 启动定时器触发 beat

celery -A django_server beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler --logfile=/var/log/dec_server/beat.log

Tips:参考链接：

到此这篇关于Django使⽤ django-celery-beat动态添加定时任务的⽂章就介绍到这了,更多相关django celery beat动态添加定时任务内容请搜索以前的⽂章或继续浏览下⾯的相关⽂章希望⼤家以后多多⽀持！