Django使用 django-celery-beat动态添加定时任务

版本信息

 # 插件安装 
 Django==2.2.2
 django-celery-beat==2.1.0
 django-redis==4.8.0
 mysqlclient==2.0.0
 django-mysql==3.2.0
 redis==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 = False
 ​
 INSTALLED_APPS = (
     ...,
     'django_celery_beat',
 )
 ​
 # celery beat配置
 # CELERY_ENABLE_UTC = False
 CELERY_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'
 ​

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_year
    django_celery_beat.models.PeriodicTasks 此模型仅用作索引以跟踪计划何时更改
   

2. 在工作目录下配置celery.py

    # -*- coding: utf-8 -*-
    # @File:    celeryc.py
    # @Content: celery定时任务配置
    ​
    ​
    import os
    from celery import Celery, platforms
    from celery.schedules import crontab
    from django.conf import settings
    ​
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_server.settings")
    app = Celery("django_server")
    app.config_from_object("django.conf:settings", namespace="CELERY")
    # 定时任务的存放位置
    app.autodiscover_tasks(["monitoring.tasks", "wechat.tasks"])
    ​
   

3. 创建tasks任务

    from celery import shared_task
    ​
    @shared_task
    def alarm_monitor_task(**kwargs):
      print("定时任务！！！")
    ​
   

4. 创建定时任务

    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)
   

1. 运行任务

    # 启动任务 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:

参考链接：

github.com/celery/djan…

pypi.org/project/dja…