摘要：的簡介是一個(gè)基于分布式消息傳輸?shù)漠惒饺蝿?wù)隊(duì)列，它專注于實(shí)時(shí)處理，同時(shí)也支持任務(wù)調(diào)度。目前支持等作為消息代理，但適用于生產(chǎn)環(huán)境的只有和官方推薦。任務(wù)處理完后保存狀態(tài)信息和結(jié)果，以供查詢。

??celery是一個(gè)基于分布式消息傳輸?shù)漠惒饺蝿?wù)隊(duì)列，它專注于實(shí)時(shí)處理，同時(shí)也支持任務(wù)調(diào)度。它的執(zhí)行單元為任務(wù)（task），利用多線程，如Eventlet，gevent等，它們能被并發(fā)地執(zhí)行在單個(gè)或多個(gè)職程服務(wù)器（worker servers）上。任務(wù)能異步執(zhí)行（后臺運(yùn)行）或同步執(zhí)行（等待任務(wù)完成）。
??在生產(chǎn)系統(tǒng)中，celery能夠一天處理上百萬的任務(wù)。它的完整架構(gòu)圖如下：

組件介紹：

Producer：調(diào)用了Celery提供的API、函數(shù)或者裝飾器而產(chǎn)生任務(wù)并交給任務(wù)隊(duì)列處理的都是任務(wù)生產(chǎn)者。

Celery Beat：任務(wù)調(diào)度器，Beat進(jìn)程會讀取配置文件的內(nèi)容，周期性地將配置中到期需要執(zhí)行的任務(wù)發(fā)送給任務(wù)隊(duì)列。

Broker：消息代理，又稱消息中間件，接受任務(wù)生產(chǎn)者發(fā)送過來的任務(wù)消息，存進(jìn)隊(duì)列再按序分發(fā)給任務(wù)消費(fèi)方（通常是消息隊(duì)列或者數(shù)據(jù)庫）。Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy、Zookeeper等作為消息代理，但適用于生產(chǎn)環(huán)境的只有RabbitMQ和Redis, 官方推薦 RabbitMQ。

Celery Worker：執(zhí)行任務(wù)的消費(fèi)者，通常會在多臺服務(wù)器運(yùn)行多個(gè)消費(fèi)者來提高執(zhí)行效率。

Result Backend：任務(wù)處理完后保存狀態(tài)信息和結(jié)果，以供查詢。Celery默認(rèn)已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。

??在客戶端和消費(fèi)者之間傳輸數(shù)據(jù)需要序列化和反序列化。 Celery 支出的序列化方案如下所示:

??在本文中，我們使用的celery的消息代理和后端存儲數(shù)據(jù)庫都使用redis，序列化和反序列化選擇msgpack。
??首先，我們需要安裝redis數(shù)據(jù)庫，具體的安裝方法可參考：http://www.runoob.com/redis/r... 。啟動redis，我們會看到如下界面：

在redis可視化軟件rdm中，我們看到的數(shù)據(jù)庫如下：

里面沒有任何數(shù)據(jù)。
??接著，為了能夠在python中使用celery，我們需要安裝以下模塊：

celery

redis

msgpack

這樣，我們的準(zhǔn)備工作就完畢了。

??我們創(chuàng)建的工程名稱為proj，結(jié)構(gòu)如下圖：

??首先是主程序app_test.py，代碼如下：

from celery import Celery

app = Celery("proj", include=["proj.tasks"])
app.config_from_object("proj.celeryconfig")

if __name__ == "__main__":
    app.start()

分析一下這個(gè)程序：

"from celery import Celery"是導(dǎo)入celery中的Celery類。

app是Celery類的實(shí)例，創(chuàng)建的時(shí)候添加了proj.tasks這個(gè)模塊，也就是包含了proj/tasks.py這個(gè)文件。

把Celery配置存放進(jìn)proj/celeryconfig.py文件，使用app.config_from_object加載配置。

??接著是任務(wù)函數(shù)文件tasks.py，代碼如下：

import time
from proj.app_test import app

@app.task
def add(x, y):
    time.sleep(1)
    return x + y

tasks.py只有一個(gè)任務(wù)函數(shù)add，讓它生效的最直接的方法就是添加app.task這個(gè)裝飾器。add的功能是先休眠一秒，然后返回兩個(gè)數(shù)的和。

??接著是配置文件celeryconfig.py，代碼如下：

BROKER_URL = "redis://localhost" # 使用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 # 任務(wù)過期時(shí)間

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

??最后是調(diào)用文件diaoyong.py，代碼如下：

from proj.tasks import add
import time

t1 = time.time()

r1 = add.delay(1, 2)
r2 = add.delay(2, 4)
r3 = add.delay(3, 6)
r4 = add.delay(4, 8)
r5 = add.delay(5, 10)

r_list = [r1, r2, r3, r4, r5]
for r in r_list:
    while not r.ready():
        pass
    print(r.result)

t2 = time.time()

print("共耗時(shí)：%s" % str(t2-t1))

在這個(gè)程序中，我們調(diào)用了add函數(shù)五次，delay()用來調(diào)用任務(wù)。

??到此為止，我們已經(jīng)理解了整個(gè)項(xiàng)目的結(jié)構(gòu)與代碼。
??接下來，我們嘗試著把這個(gè)項(xiàng)目運(yùn)行起來。
??首先，我們需要啟動redis。接著，切換至proj項(xiàng)目所在目錄，并運(yùn)行命令：

celery -A proj.app_test worker -l info

界面如下：

然后，我們運(yùn)行diaoyong.py，輸出的結(jié)果如下：

3
6
9
12
15
共耗時(shí)：1.1370790004730225

后臺輸出如下：

接著，我們看一下rdm中的數(shù)據(jù)：

至此，我們已經(jīng)成功運(yùn)行了這個(gè)項(xiàng)目。
??下面，我們嘗試著對這個(gè)運(yùn)行結(jié)果做些分析。首先，我們一次性調(diào)用了五次add函數(shù)，但是運(yùn)行的總時(shí)間才1秒多。這是celery異步運(yùn)行的結(jié)果，如果是同步運(yùn)行，那么，至少需要5秒多，因?yàn)槊空{(diào)用add函數(shù)一次，就會休眠一秒。這就是celery的強(qiáng)大之處。
??從后臺輸出可以看到，程序會先將任務(wù)分發(fā)出來，每個(gè)任務(wù)一個(gè)ID，在后臺統(tǒng)一處理，處理完后會有相應(yīng)的結(jié)果返回，同時(shí)該結(jié)果也會儲存之后臺數(shù)據(jù)庫。可以利用ready()判斷任務(wù)是否執(zhí)行完畢，再用result獲取任務(wù)的結(jié)果。
??本文項(xiàng)目的github地址為：https://github.com/percent4/c... 。
??本次分享到此結(jié)束，感謝閱讀~
??注意：本人現(xiàn)已開通微信公眾號： Python爬蟲與算法（微信號為：easy_web_scrape）， 歡迎大家關(guān)注哦~~

Celery 初步：http://docs.jinkan.org/docs/c...

使用Celery：https://zhuanlan.zhihu.com/p/...

異步神器celery：https://www.jianshu.com/p/9be...