怎么用django-celery-beat搭建定時任務

本篇內容主要講解“怎么用django-celery-beat搭建定時任務”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“怎么用django-celery-beat搭建定時任務”吧!

一、創建django項目和app

1、安裝定時任務第三方包

pip install django-celery-beat # 插件用來動態配置定時任務,一般會配合 django_celery_results 一起使用，所以一起安裝 django_celery_results

pip install django_celery_results
pip install eventlet # windows下運行celery 4以后版本，還需額外安裝eventlet庫

2、創建django項目并創建一個使用定時任務的app

1.1創建django項目并創建app

創建的過程省略，不在這里展開，需要注意的是setting文件注冊app的配置如下：

INSTALLED_APPS = [
    ......
    'myapp',  # 剛創建的使用定時任務的app
    'django_celery_beat',  # 插件用來動態配置定時任務，只要進行了第一步pip安裝就可以直接注冊了
    'django_celery_results',
]

1.2 創建定時任務數據庫
依次執行： python manage.py makemigrations 和 python manage.py migrate
打開數據庫發現，自動創建了一些表如下：

這些都是定時任務需要的表格，自動創建不需要手動管理

django_celery_beat.models.ClockedSchedule # 特定時刻任務
django_celery_beat.models.CrontabSchedule # 特定時間表任務，例如每周1運行的計劃
django_celery_beat.models.IntervalSchedule # 以特定間隔（例如，每5秒）運行的計劃。
django_celery_beat.models.PeriodicTask # 此模型定義要運行的單個周期性任務。
django_celery_beat.models.PeriodicTasks # 此模型僅用作索引以跟蹤計劃何時更改
django_celery_beat.models.SolarSchedule # 定制任務

如果安裝注冊了django_celery_results 還會有另外三個表:

2、新建一個celery.py文件

文件的作用是指定django環境、創建Celery app和指定Celery配置文件的啟動位置，類似與wsgi.py或asgi.py，因此也建議文件創建位置與wsgi.py或asgi.py同級。
文件內容如下：

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery, platforms

# 設置django環境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'lc_manage.settings.settings')

# 創建一個Celery app
app = Celery('djangotask')
platforms.C_FORCE_ROOT = True   # 如果配置沒有生效需要在啟動時設置  export C_FORCE_ROOT="true"

#  使用CELERY_ 作為前綴，在celeryconfig.py中寫配置
app.config_from_object('lc_manage.celeryconfig', namespace="CELERY")
# app.config_from_object('lc_manage.celeryconfig')

# 發現任務文件每個app下的tasks.py
app.autodiscover_tasks()

3、創建配置文件config.py

這個文件里的內容可以寫到項目的 settings.py里面，因為上面的celery.py 中可以指定配置文件位置；不過內容比較多，還是建議單獨創建一個配置文件celeryconfig.py，可以與celery.py 同級目錄，文件內容如下：

from __future__ import absolute_import

# broker 設置 指定中間代理人將任務存到哪里,這里是redis的11號庫
CELERY_BROKER_URL = 'redis://:123456@127.0.0.1:6379/11'
# 指定 Backend 儲存結果的地方，可以使用django數據庫（django-db），也可以使用redis，
# 使用django數據庫（django-db），以后運行worker就會保存到數據庫中，可以通過ORM進行訪問
# CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'
CELERY_RESULT_BACKEND = 'django-db'

# 使用django_celery_beat插件用來動態配置任務
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

# 指定時區，默認是 UTC
CELERY_TIMEZONE = 'Asia/Shanghai'

# celery 序列化與反序列化配置
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'
CELERY_ACCEPT_CONTENT = ['pickle', 'json']
CELERY_TASK_IGNORE_RESULT = True

# 有些情況下可以防止死鎖  非常重要！
CELERYD_FORCE_EXECV = True
# 為存儲結果設置過期日期，默認1天過期。如果beat開啟，Celery每天會自動清除。
# 設為0，存儲結果永不過期
# CELERY_RESULT_EXPIRES = xx
# CELERY_TASK_RESULT_EXPIRES = 60*60*24  # 后端存儲的任務超過一天時，自動刪除數據庫中的任務數據，單位秒
CELERY_MAX_TASKS_PER_CHILD = 1000  # 每個worker執行1000次任務后，自動重啟worker，防止任務占用太多內存導致內存泄漏
# 禁用所有速度限制，如果網絡資源有限，不建議開足馬力。
CELERY_DISABLE_RATE_LIMITS = True
# celery beat配置（周期性任務設置）
CELERY_ENABLE_UTC = False

# 官方用來修復CELERY_ENABLE_UTC=False and USE_TZ = False 時時間比較錯誤的問題；
# 詳情見：https://github.com/celery/django-celery-beat/pull/216/files
DJANGO_CELERY_BEAT_TZ_AWARE = False

這里需要注意的是，如果在celery.py中配置指定了confiig配置文件使用CELERY前綴：app.config_from_object(‘lc_manage.celeryconfig’, namespace=“CELERY”)，那么celeryconfig.py配置文件的參數都應加:CELERY_，當然你也可以不用第二個參數，namespace=“CELERY"寫成app.config_from_object(‘lc_manage.celeryconfig’”)， 那么celeryconfig.py中就不需要加CELERY_ 前綴，注意一定要統一！！！否則可能 會報錯：

consumer: Cannot connect to amqp://guest:**@127.0.0.1:5672//: [Errno 61] Connection refused.

4、加載celery.py

我們自己創建的celery.py雖然與wsgi.py 或者 asgi.py等同級，但是 不會像他們一樣自動加載，需要我們通過本級文件下的__init__.py 把celery.py 加載進來，打開__init__.py文件，添加如下內容：

 from __future__ import absolute_import, unicode_literals
 from .celery import app as celery_app
# 使得django啟動時加載celery的app
__all__ = ('celery_app',)

5、創建定時任務執行內容

經過上面的配置，django-celery_beta 會自動去掃描每個app目錄下是否有 tasks.py 文件，需要創建定時任務的app下我們可以手動創建tasks.py，定時任務就寫在這個文件上:

from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def add(x, y):
    print("x + y  =  ", x + y)
    return x + y

@shared_task
def mul(x, y):
    print("x * y  =  ", x * y)
    return x * y

二、定時器創建和定時任務添加

1、時間和周期控制：IntervalSchedule、ClockedSchedule和CrontabSchedule

其他帖子都把上面三個稱為定時任務與PeriodicTask放一起結束，但是個人理解以上三個都是定時任務時控制時間和周期執行的控制器，并非創建定時任務，真正創建定時任務的只有PeriodicTask，所以這里個人把這三個稱為定時任務的“時間和頻率控制器”。

• IntervalSchedule 按時間間隔頻率執行定時任務的控制器，(例：每間隔1H/1M/…執行一次)

• ClockedSchedule 指定某個時刻執行定時任務的控制器， （例：2018年8月8號 8:00這個時刻執行）

• CrontabSchedule 指定某個時間執行定時任務的控制器 (例：每年的12月星期一的8:30)

導入這四個模塊：

from django_celery_beat.models import CrontabSchedule, PeriodicTask, IntervalSchedule,ClockedSchedule

1.1 IntervalSchedule 時間間隔控制器

參數：every 間隔數，period 間隔單位

schedule, created = IntervalSchedule.objects.update_or_create(
    every=1,
    period=IntervalSchedule.MINUTES)   # 按分鐘間隔執行

第二個參數可選

• IntervalSchedule.DAYS 固定間隔天數

• IntervalSchedule.HOURS 固定間隔小時數

• IntervalSchedule.MINUTES 固定間隔分鐘數

• IntervalSchedule.SECONDS 固定間隔秒數

• IntervalSchedule.MICROSECONDS 固定間隔微秒

返回值可直接解包，其實只有第一個參數schedule有用，在PeriodicTask創建定時任務時作為時間和周期控制參數傳入。解包獲得的第二個數據created 可以直接用下劃線取代

1.2 ClockedSchedule特定時刻定時器

參數：clocked_time 指定時間

clocked, _ = ClockedSchedule.objects.update_or_create(
    clocked_time =datetime.strptime("2020-08-18 16:58:46","%Y-%m-%d %H:%M:%S"))   # 按指定時間執行

這里第二個參數直接用下劃線取代。特定時刻控制一般時執行一次，適合在view中調用執行一次性計劃。

1.3、周期性任務 CrontabSchedule

參數：

• month_of_year # 幾月執行

• day_of_month # 幾號執行

• day_of_week # 周幾執行

• hour # 幾點執行

• minute # 幾分執行

• timezone # 時區

crontab, _ = CrontabSchedule.objects.update_or_create(
    minute="00",
    hour="23",   
    day_of_week="*",   # 周幾執行
    day_of_month='1',  # 幾點執行
    month_of_year='*',    
    timezone=pytz.timezone("Asia/Shanghai"),
)

上面的星號代表不使用，上面的配置即： 每月1日的23點執行一次。如果day_of_month=“*”則代表每天23點執行。

2、動態添加任務 PeriodicTask

參數：

• name：任務名

• task：指定的任務

• interval：時間間隔

• crontab：時間控制器

• clocked：指定時刻控制器

• expires：有效日期

• one_off：啟用狀態(如果為True，調度將只運行該任務一次)

• start_time：開始時間

• enabled：啟用

• last_run_at：最后運行時間

• total_run_count：運行總次數

• date_changed：最后的更改時間

• description：描述

• args: 傳參

注意：

1、interval、crontab、clocked三選一，不可同時使用。參數值分別對應interval：schedule(IntervalSchedule的第一個返回值)；crontab：crontab（CrontabSchedule的第一個返回值）；clocked:clocked（ClockedSchedule返回值）
2、name ：任務名，確保其唯一性！！！！
3、task：后面是以字符串形式調用定時任務的具體工作內容函數，即第一項的第5條用@shared_task裝飾器創建的方法。
4、enabled：是否啟用，這里動態創建的話一般設為true
5、args: 傳參（task中指定的任務需要傳參時使用），注意需要json序列化 json.dumps(…)

其他參數均為非必填項。

PeriodicTask.objects.update_or_create(
    name=working_point_record_id + 'working_time_1',
    defaults={
        "task": "apps.tasks.add",
        "crontab": crontab,
        "enabled": True,
        "args": json.dumps([working_point_record_id])
    },

4、封裝方法

一般情況下，可以把IntervalSchedule、ClockedSchedule和CrontabSchedule根據業務需求單獨封裝一個文件，而PeriodicTask.objects直接在對應view視圖中調用即可。

三、啟動定時任務beat

定時任務是獨立與django項目運行的，django只是定時任務的入口和操作數據庫的入口，而這前提是django-celery-beat 已經獨立啟動，django-celery-beat的啟動分兩步，一是生產者單獨啟動(beat)，而是工作者單獨啟動(worker)，這里啟動順序需要注意一點，建議先啟動worker 再啟動beat ，曾經遇到先啟動beat有可能beat會啟動失敗。

1、啟動工作者worker

此時仍然需要重新打開一個命令窗口，進入django項目的根目錄（manage.py同級目錄）下：

# Linux下測試，啟動Celery
 Celery -A 【項目名稱】worker -l info  
 
 # Windows下測試，啟動Celery
 Celery -A 【項目名稱】worker -l info -P eventlet
 
 # 如果Windows下Celery不工作，輸入如下命令
 Celery -A 【項目名稱】worker -l info --pool=solo

2、啟動生產者beat

beat 是一個生產者角色，是單獨運行，生產者完全不依賴django,需要與django在一個不同的命令窗口運行，但啟動時需要先進入django項目的根目錄，即與manage.py在同一目錄下：

celery  -A 【項目名稱】 beat -l info

四、定時任務創建

這里要劃重點了，其實看完上面的已經可以使用了，下面的屬于優化選擇性理解，看個人理解能力和需求：

一般情況下，定時器創建和定時任務添加 PeriodicTask會在view視圖中創建，但是如果view視圖比較頻繁，那么每次執行view都創建一個定時任務的話會有無數個，會比較占用內存資源，當然可以在celeryconfig.py設置執行多少次后重啟任務，高并發下view會不會導致頻繁重啟，這里需要根據業務邏輯把@shared_task包裹下的任務處理方法做成批量處理放到單獨文件中（不需要在view視圖中調用，定時調用批處理，那么就是一個任務批量處理數據而已），在間隔固定時間下執行，單純語言難以表達清楚，理解的就理解了，不理解的就慢慢體會吧。這里需要畫重點的是：如果你理解了單個文件寫@shared_task批處理方法，那么你的問題重點就是：我如何啟動它呢？難道每次在啟動beat后在命令行啟動嗎？雖然也可以，但是如果項目需要創建的定時任務很多怎么辦，不用多，幾百行可以了，每次在命令行復制粘貼執行命令 嗎？當然不要，其實直接創建一個單獨的python文件，然后再根目錄下的url中導入即可，因為url在項目啟動的時候會自動加載一次，也就相當于啟動django項目的時候就自動創建了一次定時任務！

到此，相信大家對“怎么用django-celery-beat搭建定時任務”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！