celery非同步任務框架

首頁>技術>Python學習2020-09-03 02:10

celery非同步任務框架

Celery一、官方二、Celery非同步任務框架Celery架構圖訊息中介軟體任務執行單元任務結果儲存三、使用場景四、Celery的安裝配置五、兩種celery任務結構：提倡用包管理，結構更清晰七、Celery執行非同步任務包架構封裝八、基本使用celery.py 基本配置tasks.py 新增任務add_task.py 新增立即、延遲任務get_result.py 獲取結果九、高階使用celery.py 定時任務配置(迴圈的)tasks.pyget_result.py十、django中使用(更新輪播圖案例)redis的配置介面快取views.py啟動服務celery.pytasks.pyCelery一、官方

Celery 官網：http://www.celeryproject.org/

Celery 官方文件英文版：http://docs.celeryproject.org/en/latest/index.html

Celery 官方文件中文版：http://docs.jinkan.org/docs/celery/

二、Celery非同步任務框架

&#34;&#34;&#34;1）可以不依賴任何伺服器，通過自身命令，啟動服務(內部支援socket)2）celery服務為為其他專案服務提供非同步解決任務需求的注：會有兩個服務同時執行，一個是專案服務，一個是celery服務，專案服務將需要非同步處理的任務交給celery服務，celery就會在需要時非同步完成專案的需求人是一個獨立執行的服務 | 醫院也是一個獨立執行的服務\t正常情況下，人可以完成所有健康情況的動作，不需要醫院的參與；但當人生病時，就會被醫院接收，解決人生病問題\t人生病的處理方案交給醫院來解決，所有人不生病時，醫院獨立執行，人生病時，醫院就來解決人生病的需求&#34;&#34;&#34;

Celery架構圖

Celery的架構由三部分組成，訊息中介軟體（message broker）、任務執行單元（worker）和任務執行結果儲存（task result store）組成。

訊息中介軟體

Celery本身不提供訊息服務，但是可以方便的和第三方提供的訊息中介軟體整合。包括，RabbitMQ, Redis等等

任務執行單元

Worker是Celery提供的任務執行的單元，worker併發的執行在分散式的系統節點中。

任務結果儲存

Task result store用來儲存Worker執行的任務的結果，Celery支援以不同方式儲存任務的結果，包括AMQP, redis等

三、使用場景

非同步執行：解決耗時任務

延遲執行：解決延遲任務

定時執行：解決週期(週期)任務

四、Celery的安裝配置

pip install celery

訊息中介軟體：RabbitMQ/Redis

app=Celery('任務名', broker='xxx', backend='xxx')

五、兩種celery任務結構：提倡用包管理，結構更清晰

# 如果 Celery物件:Celery(...) 是放在一個模組下的# 1）終端切換到該模組所在資料夾位置：scripts# 2）執行啟動worker的命令：celery worker -A 模組名 -l info -P eventlet# 注：windows系統需要eventlet支援，Linux與MacOS直接執行：celery worker -A 模組名 -l info# 注：模組名隨意# 如果 Celery物件:Celery(...) 是放在一個包下的# 1）必須在這個包下建一個celery.py的檔案，將Celery(...)產生物件的語句放在該檔案中# 2）執行啟動worker的命令：celery worker -A 包名 -l info -P eventlet# 注：windows系統需要eventlet支援，Linux與MacOS直接執行：celery worker -A 模組名 -l info# 注：包名隨意

放在根目錄下就行：

七、Celery執行非同步任務包架構封裝

project ├── celery_task \t# celery包 │ ├── __init__.py # 包檔案 │ ├── celery.py # celery連線和配置相關檔案，且名字必須交celery.py │ └── tasks.py # 所有任務函式 ├── add_task.py \t# 新增任務 └── get_result.py # 獲取結果

八、基本使用celery.py 基本配置

# 1）建立app + 任務# 2）啟動celery(app)服務：# 非windows# 命令：celery worker -A celery_task -l info# windows：# pip3 install eventlet# celery worker -A celery_task -l info -P eventlet# 3）新增任務：手動新增，要自定義新增任務的指令碼，右鍵執行指令碼# 4）獲取結果：手動獲取，要自定義獲取任務的指令碼，右鍵執行指令碼from celery import Celery# 無密碼broker = &#39;redis://127.0.0.1:6379/1&#39; backend = &#39;redis://127.0.0.1:6379/2&#39;# 有密碼：broker = &#39;redis://:[email protected]:6379/1&#39;backend = &#39;redis://:[email protected]:6379/2&#39;app = Celery(broker=broker, backend=backend, include=[&#39;celery_task.tasks&#39;])&#39;&#39;&#39;broker : 任務倉庫backend ： 任務結果倉庫include ：任務(函式)所在檔案&#39;&#39;&#39;

tasks.py 新增任務

from .celery import [email protected] add(n1,n2): res = n1+n2 print(&#39;n1+n2 = %s&#39; % res) return [email protected] low(n1,n2): res = n1-n2 print(&#39;n1-n2 = %s&#39; % res) return res

add_task.py 新增立即、延遲任務

from celery_task import tasks# delay ：新增立即任務# apply_async ： 新增延遲任務# eta ： 執行的utc時間# 新增立即執行任務t1 = tasks.add.delay(10, 20)t2 = tasks.low.delay(100, 50)print(t1.id)# 新增延遲任務from celery_package.tasks import jumpfrom datetime import datetime,timedelta# 秒def eta_second(second): ctime = datetime.now() # 當前時間 utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) # 當前UTC時間 time_delay = timedelta(seconds=second) # 秒 return utc_ctime + time_delay # 當前時間+往後延遲的秒# 天def eta_days(days): ctime = datetime.now() # 當前時間 utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) # 當前UTC時間 time_delay = timedelta(days=days) # 天 return utc_ctime + time_delay # 當前時間+往後延遲的天jump.apply_async(args=(20,5), eta=eta_second(10)) # 10秒後執行jump.apply_async(args=(20,5), eta=eta_days(1)) # 1天后執行

get_result.py 獲取結果

from celery_task.celery import appfrom celery.result import AsyncResultid = &#39;21325a40-9d32-44b5-a701-9a31cc3c74b5&#39;if __name__ == &#39;__main__&#39;: async = AsyncResult(id=id, app=app) if async.successful(): result = async.get() print(result) elif async.failed(): print(&#39;任務失敗&#39;) elif async.status == &#39;PENDING&#39;: print(&#39;任務等待中被執行&#39;) elif async.status == &#39;RETRY&#39;: print(&#39;任務異常後正在重試&#39;) elif async.status == &#39;STARTED&#39;: print(&#39;任務已經開始被執行&#39;)

九、高階使用celery.py 定時任務配置(迴圈的)

特點：

新增任務的終端關閉之後，停止新增

celery服務端關閉後，把關閉之後未執行的任務都執行一遍，然後繼續接收任務

# 1）建立app + 任務# 2）啟動celery(app)服務：# 注）：-A 表示相對路徑，所以一定先進入celery_task所在包\t -l 表示列印到日誌 info 級別# 非windows# 命令：celery worker -A celery_task -l info# windows：# pip3 install eventlet# celery worker -A celery_task -l info -P eventlet# 3）新增任務：自動新增任務，所以要啟動一個新增任務的服務# 命令：celery beat -A celery_task -l info# 4）獲取結果from celery import Celery# 無密碼broker = &#39;redis://127.0.0.1:6379/1&#39; backend = &#39;redis://127.0.0.1:6379/2&#39;# 有密碼：broker = &#39;redis://:[email protected]:6379/1&#39;backend = &#39;redis://:[email protected]:6379/2&#39;app = Celery(broker=broker, backend=backend, include=[&#39;celery_task.tasks&#39;])# 時區app.conf.timezone = &#39;Asia/Shanghai&#39;# 是否使用UTCapp.conf.enable_utc = False# 自動任務的定時配置from datetime import timedeltafrom celery.schedules import crontabapp.conf.beat_schedule = { # 定時任務名字 &#39;fall_task&#39;: { &#39;task&#39;: &#39;celery_task.tasks.fall&#39;, &#39;args&#39;:(30,20), &#39;schedule&#39;: timedelta(seconds=3), # 3秒後執行 # &#39;schedule&#39;: crontab(hour=8, day_of_week=1), # 每週一早八點 }}&#39;&#39;&#39;fall_task：任務名自定義task:任務來源args:任務引數schedule：定時時間&#39;&#39;&#39;&#39;schedule&#39;: crontab(hour=8, day_of_week=1), # 每週一早八點&#39;&#39;&#39;minute : 分鐘hour ：小時day_of_week ：禮拜day_of_month:月month_of_year:年&#39;&#39;&#39;

tasks.py

from .celery import [email protected] fall(n1,n2): res = n1/n2 print(&#39;n1 /n2 = %s&#39; % res) return res

get_result.py

from celery_task.celery import appfrom celery.result import AsyncResultid = &#39;21325a40-9d32-44b5-a701-9a31cc3c74b5&#39;if __name__ == &#39;__main__&#39;: async = AsyncResult(id=id, app=app) if async.successful(): result = async.get() print(result) elif async.failed(): print(&#39;任務失敗&#39;) elif async.status == &#39;PENDING&#39;: print(&#39;任務等待中被執行&#39;) elif async.status == &#39;RETRY&#39;: print(&#39;任務異常後正在重試&#39;) elif async.status == &#39;STARTED&#39;: print(&#39;任務已經開始被執行&#39;)

十、django中使用(更新輪播圖案例)

最終達到的效果：根據定時任務來更新redis中的快取。使用者獲取資源都是從redis快取中獲取。避免了資料庫的壓力

redis的配置

dev.py

啟動服務

&#39;&#39;&#39;1):先切換到celery_task所在的同級目錄(一般為根目錄下)2):開一個終端(啟動服務)：\t\tcelery worker -A celery_task -l info -P eventlet3):再開一個終端(新增任務):\tcelery beat -A celery_task -l info&#39;&#39;&#39;# 注）：-A 表示相對路徑，所以一定先進入celery_task所在包\t -l 表示列印到日誌 info 級別

celery.py

&#34;&#34;&#34;celery框架django專案工作流程1）載入django配置環境2）建立Celery框架物件app，配置broker和backend，得到的app就是worker3）給worker對應的app新增可處理的任務函式，用include配置給worker的app4）完成提供的任務的定時配置app.conf.beat_schedule5）啟動celery服務，執行worker，執行任務6）啟動beat服務，執行beat，新增任務重點：由於採用了django的反射機制，使用celery.py所在的celery_task包必須放置專案的根目錄下&#34;&#34;&#34;# 一、載入django配置環境import osos.environ.setdefault(&#34;DJANGO_SETTINGS_MODULE&#34;, &#34;luffyapi.settings.dev&#34;)# 二、載入celery配置環境from celery import Celerybroker = &#39;redis://:[email protected]:6379/1&#39;backend = &#39;redis://:[email protected]:6379/2&#39;# workerapp = Celery(broker=broker, backend=backend, include=[&#39;celery_task.tasks&#39;]) # 外面的包名和檔名，一般都是固定# 時區app.conf.timezone = &#39;Asia/Shanghai&#39;# 是否使用UTCapp.conf.enable_utc = False# 任務的定時配置from datetime import timedeltafrom celery.schedules import crontabapp.conf.beat_schedule = { # 定時任務名字 &#39;update_banner_cache&#39;: { &#39;task&#39;: &#39;celery_task.tasks.update_banner_list&#39;, &#39;args&#39;: (), &#39;schedule&#39;: timedelta(seconds=10), # 3秒一次 # &#39;schedule&#39;: crontab(hour=8, day_of_week=1), # 每週一早八點 # &#39;schedule&#39;: crontab(minute=0, day_of_week=1), # 每週一早八點 }}&#39;&#39;&#39;minute : 分鐘hour ：小時day_of_week ：禮拜day_of_month:月month_of_year:年&#39;&#39;&#39;&#39;&#39;&#39;fall_task：任務名自定義task:任務來源args:任務引數schedule：定時時間(秒)&#39;&#39;&#39;

tasks.py

from .celery import appfrom django.core.cache import cachefrom home import models, serializersfrom django.conf import [email protected] update_banner_list(): queryset = models.Banner.objects.filter(is_delete=False, is_show=True).order_by(&#39;-orders&#39;)[:settings.BANNER_COUNT] banner_list = serializers.BannerSerializer(queryset, many=True).data # 拿不到request物件，所以頭像的連線base_url要自己組裝 for banner in banner_list: banner[&#39;image&#39;] = &#39;http://127.0.0.1:8000%s&#39; % banner[&#39;image&#39;] cache.set(&#39;banner_list&#39;, banner_list, 86400) return True

選擇了IT，必定終身學習

出處：http://dwz.date/aNfM

176

Redis

Windows

最新評論

劇多