Python celery原理及運行流程解析
celery簡介
celery是一個基于分布式消息傳輸的異步任務隊列,它專注于實時處理,同時也支持任務調度。它的執行單元為任務(task),利用多線程,如Eventlet,gevent等,它們能被并發地執行在單個或多個職程服務器(worker servers)上。任務能異步執行(后臺運行)或同步執行(等待任務完成)。
在生產系統中,celery能夠一天處理上百萬的任務。它的完整架構圖如下:
組件介紹:
Producer:調用了Celery提供的API、函數或者裝飾器而產生任務并交給任務隊列處理的都是任務生產者。 Celery Beat:任務調度器,Beat進程會讀取配置文件的內容,周期性地將配置中到期需要執行的任務發送給任務隊列。 Broker:消息代理,又稱消息中間件,接受任務生產者發送過來的任務消息,存進隊列再按序分發給任務消費方(通常是消息隊列或者數據庫)。Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy、Zookeeper等作為消息代理,但適用于生產環境的只有RabbitMQ和Redis, 官方推薦 RabbitMQ。 Celery Worker:執行任務的消費者,通常會在多臺服務器運行多個消費者來提高執行效率。 Result Backend:任務處理完后保存狀態信息和結果,以供查詢。Celery默認已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。工作原理
它的基本工作就是管理分配任務到不同的服務器,并且取得結果。至于說服務器之間是如何進行通信的?這個Celery本身不能解決。所以,RabbitMQ作為一個消息隊列管理工具被引入到和Celery集成,負責處理服務器之間的通信任務。和rabbitmq的關系只是在于,celery沒有消息存儲功能,他需要介質,比如rabbitmq、redis、mysql、mongodb 都是可以的。推薦使用rabbitmq,他的速度和可用性都很高。
Celery安裝及使用
1、安裝celery
pip install celery
2、查看完整可用命令選項
celery worker --help
3、創建一個工程項目project,然后再項目內創建一個celery_tasks異步任務列表。如圖:
4、首先是celery_tasks異步任務主程序main.py,代碼如下:
from celery import Celery# 生成celery應用celery_app = Celery('caicai')# 加載配置文件celery_app.config_from_object(’celery_tasks.config’)# 注冊任務celery_app.autodiscover_tasks([’celery_tasks.email’]) # 注意:傳遞的參數是任務列表
分析一下這個程序:
'from celery import Celery'是導入celery中的Celery類。celery_app celery_app是Celery類的實例。 把Celery配置存放進project/config.py文件,使用celery_app.config_from_object加載配置。 將任務注冊到應用中5、接著是配置文件config.py,代碼如下:
BROKER_URL = ’redis://localhost:6379/1’ # 使用Redis作為消息代理CELERY_RESULT_BACKEND = ’redis://localhost:6379/0’ # 把任務結果存在了Redis# CELERY_TASK_SERIALIZER = ’msgpack’ # 任務序列化和反序列化使用msgpack方案CELERY_RESULT_SERIALIZER = ’json’ # 讀取任務結果一般性能要求不高,所以使用了可讀性更好的JSONCELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # celery任務結果有效期CELERY_ACCEPT_CONTENT = [’json’, ’msgpack’] # 指定接受的內容類型CELERY_TIMEZONE = ’Asia/Shanghai’ # celery使用的時區CELERY_ENABLE_UTC = True # 啟動時區設置CELERYD_LOG_FILE = '/var/log/celery/celery.log' # celery日志存儲位置
6、創建email目錄,目錄下創建tesks.py文件用來編寫發送郵件的代碼,代碼如下:
import timefrom celery_tasks.main import celery_app@celery_app.task(name=’seed_email’) # 添加celery_app.task這個裝飾器,指定該任務的任務名name=’seed_email’def seed(): time.sleep(1) return '我將發送郵件'
7、在項目app.py中,采用delay()用來調用任務。
from celery_tasks.email.tasks import seedseed.delay()seed.delay()seed.delay()seed.delay()seed.delay()
8、項目運行
首先,我們需要啟動redis。接著,切換至proj項目所在目錄,并運行命令:
celery -A celery_tasks.main worker -l info
界面如下:
然后,我們運行app.py,app.py調用添加異步任務,輸出的結果如下:
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持好吧啦網。
相關文章: