celery簡介

celery是一個基于分布式消息傳輸?shù)漠惒饺蝿?wù)隊列，它專注于實時處理，同時也支持任務(wù)調(diào)度。它的執(zhí)行單元為任務(wù)（task），利用多線程，如Eventlet，gevent等，它們能被并發(fā)地執(zhí)行在單個或多個職程服務(wù)器（worker servers）上。任務(wù)能異步執(zhí)行（后臺運行）或同步執(zhí)行（等待任務(wù)完成）。

在生產(chǎn)系統(tǒng)中，celery能夠一天處理上百萬的任務(wù)。它的完整架構(gòu)圖如下：

組件介紹：

• Producer：調(diào)用了Celery提供的API、函數(shù)或者裝飾器而產(chǎn)生任務(wù)并交給任務(wù)隊列處理的都是任務(wù)生產(chǎn)者。
• Celery Beat：任務(wù)調(diào)度器，Beat進程會讀取配置文件的內(nèi)容，周期性地將配置中到期需要執(zhí)行的任務(wù)發(fā)送給任務(wù)隊列。
• Broker：消息代理，又稱消息中間件，接受任務(wù)生產(chǎn)者發(fā)送過來的任務(wù)消息，存進隊列再按序分發(fā)給任務(wù)消費方（通常是消息隊列或者數(shù)據(jù)庫）。Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy、Zookeeper等作為消息代理，但適用于生產(chǎn)環(huán)境的只有RabbitMQ和Redis, 官方推薦 RabbitMQ。
• Celery Worker：執(zhí)行任務(wù)的消費者，通常會在多臺服務(wù)器運行多個消費者來提高執(zhí)行效率。
• Result Backend：任務(wù)處理完后保存狀態(tài)信息和結(jié)果，以供查詢。Celery默認已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。
  

工作原理

它的基本工作就是管理分配任務(wù)到不同的服務(wù)器，并且取得結(jié)果。至于說服務(wù)器之間是如何進行通信的？這個Celery本身不能解決。所以，RabbitMQ作為一個消息隊列管理工具被引入到和Celery集成，負責處理服務(wù)器之間的通信任務(wù)。和rabbitmq的關(guān)系只是在于，celery沒有消息存儲功能，他需要介質(zhì)，比如rabbitmq、redis、mysql、mongodb 都是可以的。推薦使用rabbitmq，他的速度和可用性都很高。

Celery安裝及使用

1、安裝celery

pip install celery

2、查看完整可用命令選項

celery worker --help

3、創(chuàng)建一個工程項目project，然后再項目內(nèi)創(chuàng)建一個celery_tasks異步任務(wù)列表。如圖：

4、首先是celery_tasks異步任務(wù)主程序main.py，代碼如下：

from celery import Celery
# 生成celery應(yīng)用
celery_app = Celery("caicai")
# 加載配置文件
celery_app.config_from_object('celery_tasks.config')
# 注冊任務(wù)
celery_app.autodiscover_tasks(['celery_tasks.email']) # 注意：傳遞的參數(shù)是任務(wù)列表

分析一下這個程序：

• "from celery import Celery"是導入celery中的Celery類。celery_app
• celery_app是Celery類的實例。
• 把Celery配置存放進project/config.py文件，使用celery_app.config_from_object加載配置。
• 將任務(wù)注冊到應(yīng)用中

5、接著是配置文件config.py，代碼如下：

BROKER_URL = 'redis://localhost:6379/1' # 使用Redis作為消息代理

CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 把任務(wù)結(jié)果存在了Redis

# CELERY_TASK_SERIALIZER = 'msgpack' # 任務(wù)序列化和反序列化使用msgpack方案

CELERY_RESULT_SERIALIZER = 'json' # 讀取任務(wù)結(jié)果一般性能要求不高，所以使用了可讀性更好的JSON

CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # celery任務(wù)結(jié)果有效期

CELERY_ACCEPT_CONTENT = ['json', 'msgpack'] # 指定接受的內(nèi)容類型

CELERY_TIMEZONE = 'Asia/Shanghai'       # celery使用的時區(qū)
CELERY_ENABLE_UTC = True            # 啟動時區(qū)設(shè)置
CELERYD_LOG_FILE = "/var/log/celery/celery.log"   # celery日志存儲位置

6、創(chuàng)建email目錄，目錄下創(chuàng)建tesks.py文件用來編寫發(fā)送郵件的代碼，代碼如下：

import time
from celery_tasks.main import celery_app
@celery_app.task(name='seed_email')   # 添加celery_app.task這個裝飾器，指定該任務(wù)的任務(wù)名name='seed_email'
def seed():
  time.sleep(1)
  return "我將發(fā)送郵件"

7、在項目app.py中，采用delay()用來調(diào)用任務(wù)。

from celery_tasks.email.tasks import seed
seed.delay()
seed.delay()
seed.delay()
seed.delay()
seed.delay()

8、項目運行

　　首先，我們需要啟動redis。接著，切換至proj項目所在目錄，并運行命令：

celery -A celery_tasks.main worker -l info

　　界面如下：

然后，我們運行app.py，app.py調(diào)用添加異步任務(wù)，輸出的結(jié)果如下：

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

最新評論