目錄周期性定時任務基本使用crontab表達式自定義周期任務使用程序停止和刪除周期定時任務周期定時任務刪除，直接通過自定義任務的name字段刪除停止運行中的周期定時任務

周期性定時任務

說明：在Django中使用celery, 并結合django-celery模塊（省略安裝）。需要在配置文件中注冊：

import djcelery
INSTALLED_APPS += ('djcelery',)
CELERY_ENABLE_UTC = False
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"  # 基于數據庫模型調度

基本使用

django中使用，celery_tasks.py:

from celery.schedules import crontab
from celery.task import periodic_task

# 每天9點執行
@periodic_task(run_every=crontab(minute='0', hour='9'))
def job():
    # 業務邏輯
    print('周期定時任務！')

crontab表達式

詳見官網： https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html

例子	意義
crontab()	每分鐘執行一次。
crontab(minute=0, hour=0)	每天午夜執行。
crontab(minute=0, hour='*/3')	每三個小時執行一次：午夜、凌晨 3 點、早上 6 點、早上 9 點、中午、下午 3 點、下午 6 點、晚上 9 點。
crontab(minute=0,``hour='0,3,6,9,12,15,18,21')	和以前一樣。
crontab(minute='*/15')	每 15 分鐘執行一次。
crontab(day_of_week='sunday')	在星期日每分鐘 (!) 執行一次。
crontab(minute='*',``hour='*', day_of_week='sun')	和以前一樣。
crontab(minute='*/10',``hour='3,17,22', day_of_week='thu,fri')	每十分鐘執行一次，但僅限于周四或周五的凌晨 3-4 點、下午 5-6 點和晚上 10-11 點。
crontab(minute=0, hour='*/2,*/3')	每隔偶數小時執行一次，并且每小時可被 3 整除。這意味著：除了：1am、5am、7am、11am、1pm、5pm、7pm、11pm之外的每個小時
crontab(minute=0, hour='*/5')	執行可被 5 整除的小時。這意味著它在下午 3 點而非下午 5 點觸發（因為下午 3 點等于 24 小時時鐘值“15”，可被 5 整除）。
crontab(minute=0, hour='*/3,8-17')	每小時執行一次，可被 3 整除，并在辦公時間（上午 8 點至下午 5 點）每小時執行一次。
crontab(0, 0, day_of_month='2')	在每個月的第二天執行。
crontab(0, 0,``day_of_month='2-30/2')	在每個偶數日執行。
crontab(0, 0,``day_of_month='1-7,15-21')	在每月的第一周和第三周執行。
crontab(0, 0, day_of_month='11',``month_of_year='5')	每年5月11日執行。
crontab(0, 0,``month_of_year='*/3')	在每個季度的第一個月每天執行。

                       |

自定義周期任務

celery_tasks.py文件：定義任務加task()

from celery.task import task
@task()
def send_msg(oa_list, content, id, username, msg):
    # 邏輯
    pass

views.py寫自定義方法：

import datetime
import json
from djcelery.models import PeriodicTask, CrontabSchedule

def create_task(request):
    task_name = "test"	# 自定義，不能重復，唯一
    task = 'app01.celery_tasks.send_msg'  # 任務的注冊路徑
    task_args = [oa_list, content]
    task_kwargs = {'id': id, 'username': username, 'msg': msg}
    # 定時規則
    crontab_time = {
    	'minute': '*/1',
    	'hour': '*',
        'day_of_week': '*'
    	'day_of_month': '*',
    	'month_of_year': '*'
    }
    
    schedule = CrontabSchedule.objects.create(**crontab_time)  # 創建時間規則

    task, created = PeriodicTask.objects.get_or_create(
    	name=task_name,		# 名稱保持唯一
    	task=task,	
    	crontab=schedule,
    	enabled=True,	# 是否開啟
    	args=json.dumps(task_args)
    	kwargs=json.dumps(task_kwargs),
    	expires=datetime.datetime.now()+datetime.timedelta(days=1)	# 過期時間
    )
    if created:
        return HttpResponse("創建成功")
    else:
        return HttpResponse("創建失敗")

使用程序停止和刪除周期定時任務

周期定時任務刪除，直接通過自定義任務的name字段刪除

一般是針對自定義周期任務的刪除，寫在celery_tasks.py中的任務這樣刪除后，如果重啟celery，還會注冊進去，因為程序啟動會注冊所有app下的任務。

from djcelery.schedulers import ModelEntry, DatabaseScheduler

def delete_celery_task(task_name):
	DatabaseScheduler.delete_task(task_name)

停止運行中的周期定時任務

通過PeriodicTask表中的enabled字段來控制周期任務的啟停(不用重啟celery)。

from djcelery.models import PeriodicTask, CrontabSchedule

def celery_enabled(task_name):
    """
    停止周期任務
    :param task_name: 周期任務名
    """
    
    task = PeriodicTask.objects.filter(name=task_name).first()
    task.enabled=False	# False停止， True開啟
    task.save()    # 不要用update，任務不會停止

總結

以上是生活随笔為你收集整理的Celery周期性定时任务-periodic_task的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。