本文介紹了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.addfrom datetime import timedeltaCELERYBEAT_SCHEDULE = { 'add-every-30-seconds': { 'task': 'tasks.add', 'schedule': timedelta(seconds=30), 'args': (16, 16) },}#crontab任務(wù)#每周一7:30調(diào)用task.addfrom celery.schedules import crontabCELERYBEAT_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 datetimeimport jsonfrom djcelery import models as celery_modelsfrom 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 # 開(kāi)啟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目錄中打開(kāi)taske.py加入如下代碼
from djcelery import models as celery_modelsfrom 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í)有所幫助，也希望大家多多支持VEVB武林網(wǎng)。
新聞熱點(diǎn)
疑難解答
圖片精選
