快捷導(dǎo)航

python Celery定時(shí)任務(wù)的示例

更新時(shí)間：2018年03月13日 09:08:38 作者：sicofield

這篇文章主要介紹了python Celery定時(shí)任務(wù)的示例,小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧

本文介紹了python Celery定時(shí)任務(wù)的示例，分享給大家，具體如下：

配置

啟用Celery的定時(shí)任務(wù)需要設(shè)置CELERYBEAT_SCHEDULE 。

Celery的定時(shí)任務(wù)都由celery beat來(lái)進(jìn)行調(diào)度。celery beat默認(rèn)按照settings.py之中的時(shí)區(qū)時(shí)間來(lái)調(diào)度定時(shí)任務(wù)。

創(chuàng)建定時(shí)任務(wù)

一種創(chuàng)建定時(shí)任務(wù)的方式是配置CELERYBEAT_SCHEDULE：

#每30秒調(diào)用task.add
from datetime import timedelta

CELERYBEAT_SCHEDULE = {
  'add-every-30-seconds': {
    'task': 'tasks.add',
    'schedule': timedelta(seconds=30),
    'args': (16, 16)
  },
}

#crontab任務(wù)
#每周一7:30調(diào)用task.add
from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
  # Executes every Monday morning at 7:30 A.M
  'add-every-monday-morning': {
    'task': 'tasks.add',
    'schedule': crontab(hour=7, minute=30, day_of_week=1),
    'args': (16, 16),
  },
}

使用數(shù)據(jù)庫(kù)存儲(chǔ)定時(shí)任務(wù)

使用數(shù)據(jù)庫(kù)存儲(chǔ)定時(shí)任務(wù)需要設(shè)置CELERYBEAT_SCHEDULE如下：

import datetime
import json
from djcelery import models as celery_models
from django.utils import timezone
#創(chuàng)建任務(wù)
def create_task(name, task, task_args, crontab_time):
  '''
  name # 任務(wù)名字
  task # 執(zhí)行的任務(wù) "myapp.tasks.add"
  task_args # 任務(wù)參數(shù) {"x":1, "Y":1}

  crontab_time # 定時(shí)任務(wù)時(shí)間 格式：
  {
    'month_of_year': 9 # 月份
    'day_of_month': 5 # 日期
    'hour': 01 # 小時(shí)
    'minute':05 # 分鐘
  }
  '''

  # task任務(wù)， created是否定時(shí)創(chuàng)建
  task, created = celery_models.PeriodicTask.objects.
              get_or_create(name=name,task=task)
  # 獲取 crontab
  crontab = celery_models.CrontabSchedule.objects.
              filter(**crontab_time).first()
  if crontab is None:
  # 如果沒(méi)有就創(chuàng)建，有的話就繼續(xù)復(fù)用之前的crontab
    crontab = celery_models.CrontabSchedule.objects.
              create(**crontab_time)
  task.crontab = crontab # 設(shè)置crontab
  task.enabled = True # 開啟task
  task.kwargs = json.dumps(task_args) # 傳入task參數(shù)
  expiration = timezone.now() + datetime.timedelta(day=1)
  task.expires = expiration # 設(shè)置任務(wù)過(guò)期時(shí)間為現(xiàn)在時(shí)間的一天以后
  task.save()
  return True 

#關(guān)閉任務(wù)
def disable_task(name):
'''
關(guān)閉任務(wù)
'''
  try:
    task = celery_models.PeriodicTask.objects.get(name=name)
    task.enabled = False # 設(shè)置關(guān)閉
    task.save()
    return True
  except celery_models.PeriodicTask.DoesNotExist:
    return True

啟動(dòng)beat

執(zhí)行定時(shí)任務(wù)時(shí), Celery會(huì)通過(guò)celery beat進(jìn)程來(lái)完成。Celery beat會(huì)保持運(yùn)行, 一旦到了某一定時(shí)任務(wù)需要執(zhí)行時(shí), Celery beat便將其加入到queue中. 不像worker進(jìn)程, Celery beat只需要一個(gè)即可。而且為了避免有重復(fù)的任務(wù)被發(fā)送出去，所以Celery beat僅能有一個(gè)。

啟動(dòng)：

python manage.py celery beat --loglevel=info

其實(shí)還有一種簡(jiǎn)單的啟動(dòng)方式worker和beat一起啟動(dòng)：

python manage.py celery worker --loglevel=info --beat

定時(shí)刪除

由于很多任務(wù)都是一次執(zhí)行完就不需要，留在數(shù)據(jù)庫(kù)里就是垃圾數(shù)據(jù)了有沒(méi)有辦法清除。方法肯定有因?yàn)閐jango-celery本身就有定時(shí)任務(wù)功能我們加個(gè)任務(wù)就解決了。好我們看代碼：在django app目錄中打開taske.py加入如下代碼

from djcelery import models as celery_models
from django.utils import timezone
@task()
def delete():
  '''
  刪除任務(wù)
  從models中過(guò)濾出過(guò)期時(shí)間小于現(xiàn)在的時(shí)間然后刪除
  '''
  return celery_models.PeriodicTask.objects.filter(
              expires__lt=timezone.now()).delete()

創(chuàng)建任務(wù)腳本里設(shè)置了 expires 1天以后過(guò)期，這樣在filter的時(shí)候就能當(dāng)做條件把過(guò)期的任務(wù)找到并且刪除。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: