celery4+django2定時(shí)任務(wù)的實(shí)現(xiàn)代碼
網(wǎng)上有很多celery + django實(shí)現(xiàn)定時(shí)任務(wù)的教程,不過它們大多數(shù)是基于djcelery + celery3的;
或者是使用django_celery_beat配置較為繁瑣的。
顯然簡潔而高效才是我們最終的追求,而celery4已經(jīng)不需要額外插件即可與django結(jié)合實(shí)現(xiàn)定時(shí)任務(wù)了,原生的celery beat就可以很好的實(shí)現(xiàn)定時(shí)任務(wù)功能。
當(dāng)然使用原生方案的同時(shí)有幾點(diǎn)插件所帶來的好處被我們放棄了:
- 插件提供的定時(shí)任務(wù)管理將不在可用,當(dāng)我們只需要任務(wù)定期執(zhí)行而不需要人為調(diào)度的時(shí)候這點(diǎn)忽略不計(jì)。
- 無法高效的管理或追蹤定時(shí)任務(wù),定時(shí)任務(wù)的跟蹤其實(shí)交給日志更合理,但是對任務(wù)的修改就沒有那么方便了,不過如果不需要經(jīng)常變更/增減任務(wù)的話這點(diǎn)也在可接受范圍內(nèi)。
Celery定時(shí)任務(wù)配置
在進(jìn)行配置前先來看看項(xiàng)目結(jié)構(gòu):
. ├── linux_news │ ├── celery.py │ ├── __init__.py │ ├── settings.py │ ├── urls.py │ └── wsgi.py ├── manage.py ├── news │ ├── admin.py │ ├── apps.py │ ├── __init__.py │ ├── migrations │ ├── models │ ├── tasks.py │ ├── tests.py │ └── views └── start-celery.sh
其中news是我們的app,用于從一些rss訂閱源獲取新聞信息,linux_news則是我們的project。我們需要關(guān)心的主要是 celery.py , settings.py , tasks.py 和 start-celery.sh 。
首先是celery.py,想讓celery執(zhí)行任務(wù)就必須實(shí)例化一個(gè)celery app,并把settings.py里的配置傳入app:
import os from celery import Celery # set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'linux_news.settings') app = Celery('linux_news') # 'django.conf:settings'表示django,conf.settings也就是django項(xiàng)目的配置,celery會根據(jù)前面設(shè)置的環(huán)境變量自動查找并導(dǎo)入 # - namespace表示在settings.py中celery配置項(xiàng)的名字的統(tǒng)一前綴,這里是'CELERY_',配置項(xiàng)的名字也需要大寫 app.config_from_object('django.conf:settings', namespace='CELERY') # Load task modules from all registered Django app configs. app.autodiscover_tasks()
配置就是這么簡單,為了能在django里使用這個(gè)app,我們需要在__init__.py中導(dǎo)入它:
from .celery import app as celery_app
然后我們來看tasks.py,它應(yīng)該位于你的app目錄中,前面我們配置了自動發(fā)現(xiàn),所以celery會自動找到這些tasks,我們的tasks將寫在這一模塊中,代碼涉及了一些orm的使用,為了契合主題我做了些精簡:
from linux_news.celery import celery_app as app from .models import * import time import feedparser import pytz import html @app.task(ignore_result=True) def fetch_news(origin_name): """ fetch all news from origin_name """ origin = get_feeds_origin(origin_name) feeds = feedparser.parse(origin.feed_link) for item in feeds['entries']: entry = NewsEntry() entry.title = item.title entry.origin = origin entry.author = item.author entry.link = item.link # add timezone entry.publish_time = item.time.replace(tzinfo=pytz.utc) entry.summary = html.escape(item.summary) entry.save() @app.task(ignore_result=True) def fetch_all_news(): """ 這是我們的定時(shí)任務(wù) fetch all origins' news to db """ origins = NewsOrigin.objects.all() for origin in origins: fetch_news.delay(origin.origin_name)
tasks里是一些耗時(shí)操作,比如網(wǎng)絡(luò)IO或者數(shù)據(jù)庫讀寫,因?yàn)槲覀儾魂P(guān)心任務(wù)的返回值,所以使用 @app.task(ignore_result=True)
將其屏蔽了。
任務(wù)配置完成后我們就要配置celery了,我們選擇redis作為任務(wù)隊(duì)列,我強(qiáng)烈建議在生產(chǎn)環(huán)境中使用rabbitmq或者redis作為任務(wù)隊(duì)列或結(jié)果緩存后端,而不應(yīng)該使用關(guān)系型數(shù)據(jù)庫:
# redis REDIS_PORT = 6379 REDIS_DB = 0 # 從環(huán)境變量中取得redis服務(wù)器地址 REDIS_HOST = os.environ.get('REDIS_ADDR', 'redis') # celery settings # 這兩項(xiàng)必須設(shè)置,否則不能正常啟動celery beat CELERY_ENABLE_UTC = True CELERY_TIMEZONE = TIME_ZONE # 任務(wù)隊(duì)列配置 CELERY_BROKER_URL = f'redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}' CELERY_ACCEPT_CONTENT = ['application/json', ] CELERY_RESULT_BACKEND = f'redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}' CELERY_TASK_SERIALIZER = 'json'
然后是我們的定時(shí)任務(wù)設(shè)置:
from celery.schedules import crontab CELERY_BEAT_SCHEDULE={ 'fetch_news_every-1-hour': { 'task': 'news.tasks.fetch_all_news', 'schedule': crontab(minute=0, hour='*/1'), } }
定時(shí)任務(wù)配置對象是一個(gè)dict,由任務(wù)名和配置項(xiàng)組成,主要配置想如下:
- task:任務(wù)函數(shù)所在的模塊,模塊路徑得寫全,否則找不到將無法運(yùn)行該任務(wù)
- schedule:定時(shí)策略,一般使用 celery.schedules.crontab ,上面例子為每小時(shí)的0分執(zhí)行一次任務(wù),具體寫法與linux的crontab類似可以參考文檔說明
- args:是個(gè)元組,給出任務(wù)需要的參數(shù),如果不需要參數(shù)也可以不寫進(jìn)配置,就像例子中的一樣
- 其余配置項(xiàng)較少用,可以參考文檔
至此,配置celery beat的部分就結(jié)束了。
啟動celery beat
配置完成后只需要啟動celery了。
啟動之前配置一下環(huán)境。不要用root運(yùn)行celery!不要用root運(yùn)行celery!不要用root運(yùn)行celery!重要的事情說三遍。
start-celery.sh:
export REDIS_ADDR=127.0.0.1 celery -A linux_news worker -l info -B -f /path/to/log
-A 表示app所在的目錄,-B表示啟動celery beat運(yùn)行定時(shí)任務(wù)。
celery正常啟動后就可以通過日志來查看任務(wù)是否正常運(yùn)行了:
[2018-12-21 13:00:00,022: INFO/MainProcess] Received task: news.tasks.fetch_all_news[e4566ede-2cfa-4c19-b2f3-0c7d6c38690d]
[2018-12-21 13:00:00,046: INFO/MainProcess] Received task: news.tasks.fetch_news[583e96dc-f508-49fa-a24a-331e0c07a86b]
[2018-12-21 13:00:00,051: INFO/ForkPoolWorker-2] Task news.tasks.fetch_all_news[e4566ede-2cfa-4c19-b2f3-0c7d6c38690d] succeeded in 0.02503809699555859s: None
[2018-12-21 13:00:00,052: INFO/MainProcess] Received task: news.tasks.fetch_news[c61a3e55-dd3c-4d49-8d6d-ca9b1757db25]
[2018-12-21 13:00:00,449: INFO/ForkPoolWorker-5] Task news.tasks.fetch_news[c61a3e55-dd3c-4d49-8d6d-ca9b1757db25] succeeded in 0.39487219898728654s: None
[2018-12-21 13:00:00,606: INFO/ForkPoolWorker-3] Task news.tasks.fetch_news[583e96dc-f508-49fa-a24a-331e0c07a86b] succeeded in 0.5523456179944333s: None
以上就是celery4運(yùn)行定時(shí)任務(wù)的內(nèi)容,希望對大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。
相關(guān)文章
matplotlib實(shí)現(xiàn)自定義散點(diǎn)形狀marker的3種方法
本文主要介紹了matplotlib實(shí)現(xiàn)自定義散點(diǎn)形狀marker的3種方法,文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2021-10-10python 兩個(gè)一樣的字符串用==結(jié)果為false問題的解決
這篇文章主要介紹了python 兩個(gè)一樣的字符串用==結(jié)果為false問題的解決,具有很好的參考價(jià)值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-03-03Django實(shí)現(xiàn)后臺上傳并顯示圖片功能
這篇文章主要介紹了Django實(shí)現(xiàn)后臺上傳并顯示圖片功能,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-05-05Python中導(dǎo)入自定義模塊的幾種方法總結(jié)
這篇文章主要介紹了Python中導(dǎo)入自定義模塊的幾種方法總結(jié),具有很好的參考價(jià)值,希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-01-01解決python多線程報(bào)錯(cuò):AttributeError: Can''t pickle local object問題
這篇文章主要介紹了解決python多線程報(bào)錯(cuò):AttributeError: Can't pickle local object問題,具有很好的參考價(jià)值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-04-04Python生成任意波形并存為txt的實(shí)現(xiàn)
本文主要介紹了Python生成任意波形并存為txt的實(shí)現(xiàn),文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2021-11-11Python for循環(huán)及基礎(chǔ)用法詳解
這篇文章為大家介紹python for 循環(huán),它常用于遍歷字符串、列表、元組、字典、集合等序列類型,逐個(gè)獲取序列中的各個(gè)元素2019-11-11