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

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

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

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

配置

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

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

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

一種創(chuàng)建定時任務(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ù)庫存儲定時任務(wù)

使用數(shù)據(jù)庫存儲定時任務(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 # 定時任務(wù)時間 格式：
  {
    'month_of_year': 9 # 月份
    'day_of_month': 5 # 日期
    'hour': 01 # 小時
    'minute':05 # 分鐘
  }
  '''

  # task任務(wù)， created是否定時創(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:
  # 如果沒有就創(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ù)過期時間為現(xiàn)在時間的一天以后
  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

啟動beat

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

啟動：

python manage.py celery beat --loglevel=info

其實(shí)還有一種簡單的啟動方式worker和beat一起啟動：

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

定時刪除

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

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

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

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

您可能感興趣的文章: