Celery 和 Redis 入門

Celery 是一個普遍應用於網絡應用程序的任務處理系統。

它能夠在如下狀況下使用:

在請求響應週期中作網絡調用。服務器應當當即響應任何網絡請求。若是在請求響應週期內須要進行網絡調用，則應在週期外完成調用。例如當用戶在網站上註冊時，須要發送激活郵件。發送郵件是一種網絡調用，耗時2到3秒。用戶應該無需等待這2到3秒。所以，發送激活郵件應當在請求響應週期外完成，celery 就能實現這一點。

將一個由幾個獨立部分組成的大任務分紅多個小任務。假設你想知道臉書用戶的時間流。臉書提供不一樣的端點來獲取不一樣的數據。譬如，一個端點用以獲取用戶時間流中的圖片，一個端點獲取用戶時間流中的博文，一個端點獲得用戶的點贊信息等。若是你的函數須要和臉書的5個端點依此通訊，每一個網絡調用平均耗時2秒，你將須要10秒完成一次函數執行。可是，你能夠把這項工做分爲5個獨立的任務（你很快就會發現這很容易作到），並讓 celery 來處理這些任務。Celery 能夠並行地與這5個端點通訊，在2秒以內就能獲得全部端點的響應。

簡單的 celery 例子

假設咱們有一個函數，並傳給它一個網址列表。該函數須要獲取這些網址的響應。

沒有使用 celery

建立文件celery_blog.py：

import requests import time def func(urls): start = time.time() for url in urls: resp = requests.get(url) print resp.status_code print "It took", time.time() - start, "seconds" if __name__ == "__main__": func(["http://oneapm.com", "http://jd.com", "https://taobao.com", "http://baidu.com", "http://news.oneapm.com"])

運行：

python celery_blog.py

輸出：

使用 celery

調用 celery 的程序中最重要的組成部分爲 celery worker。

在 web 應用程序註冊的例子中，celery worker 用於發送郵件。

在臉書的例子中， celery worker 用於獲取不一樣的網址。

在咱們的 celery_blog.py 例子中， celery worker 用於獲取 URL。
celery worker 和你的應用程序/腳本是不一樣的進程，彼此獨立運行。因此你的應用程序/腳本和 celery 須要一些方法來相互溝通。

應用程序代碼須要把任務放在 celery worker 能夠取出並執行的位置。譬如，應用程序代碼將任務放在消息隊列中，celery worker 從消息隊列領取任務並執行任務。咱們將使用 Redis 做爲消息隊列。

請確認你已安裝 Redis，並能夠運行redis-server。

請確認你已安裝 celery。

修改文件 celery_blog.py，以下：

from celery import Celery app = Celery('celery_blog',broker='redis://localhost:6379/1') @app.task def fetch_url(url): resp = requests.get(url) print resp.status_code def func(urls): for url in urls: fetch_url.delay(url) if __name__ == "__main__": func(["http://oneapm.com", "http://jd.com", "https://taobao.com", "http://baidu.com", "http://news.oneapm.com"])

代碼解釋：咱們須要一個 celery 實例來啓動程序，所以建立了一個名爲 app 的 celery 實例。

在3個終端中啓動：

第一個終端，運行 redis-server

第二個終端，運行 celery worker -A celery_blog -l info -c 5 ，經過輸出能夠看到 celery 成功運行。

第三個終端，運行腳本 python celery_blog.py

能夠看到第二個終端輸出以下：

將 celery 代碼和配置保存在不一樣文件中

上面的例子中，咱們只寫了一個 celery 任務。但您的項目可能涉及多個模塊，您可能但願在不一樣的模塊中有不一樣的任務。因此讓咱們將 celery 配置移到單獨的文件中。

建立 celery_config.py

from celery import Celery app = Celery('celery_config', broker='redis://localhost:6379/0', include=['celery_blog'])

修改 celery_blog.py 代碼以下：

import requests from celery_config import app @app.task def fetch_url(url): resp = requests.get(url) print resp.status_code def func(urls): for url in urls: fetch_url.delay(url) if __name__ == "__main__": func(["http://oneapm.com", "http://jd.com", "https://taobao.com", "http://baidu.com", "http://news.oneapm.com"])

停掉以前的 celery worker ，運行：

celery worker -A celery_config -l info -c 5

打開 ipython ，運行以下命令：

In [1]: from celery_blog import func In [2]: func(["http://oneapm.com", "http://jd.com", "https://taobao.com", "http://baidu.com", "http://news.oneapm.com"])

輸出以下：

在不一樣文件中添加新的任務

您能夠添加新的模塊，並在該模塊中定義一個任務。用如下內容建立一個模塊 celery_add.py：

from celery_config import app @app.task def add(a, b): return a + b

改變 celery_config.py 包含新的模塊 celery_add.py，以下：

from celery import Celery app = Celery('celery_config', broker='redis://localhost:6379/0', include=['celery_blog', 'celery_add'])

在 ipython 輸入：

In [1]: from celery_add import add In [2]: add.delay(4, 5)

輸出以下：

在不一樣的機器上分開使用 Redis 和 celery

到目前爲止，咱們的腳本、celery worker 和 Redis 都運行在同一機器中。其實並沒有這種必要，這三者能夠運行在不一樣機器上。

celery 任務涉及到網絡請求，所以，在網絡優化的機器上使用 celery worker 能提升任務運行速度。Redis 是一種內存數據庫，在內存優化的機器上運行效率更高。

在這個例子中，我將在本地系統運行腳本和 celery worker，在分開的服務器上運行 Redis。

修改 celery_config.py 爲：

app = Celery('celery_config', broker='redis://192.168.118.148:6379/0', include=['celery_blog'])

如今我運行任何任務，腳本都將把他放在 Redis 運行的服務器（192.168.118.148）上面。

celery worker 也與 192.168.118.148 溝通,在這個 Redis 服務器上獲得任務並執行它。

注意:您必須使用正在運行 redis-server 的服務器地址。個人服務器已中止Redis,因此你將沒法鏈接到 Redis。