Python環境下安裝使用非同步任務佇列套件Celery

1.簡介

celery(芹菜)是一個非同步任務佇列/基於分散式訊息傳遞的作業佇列。它側重於即時操作，但對調度支援也很好。
celery用於生產系統每天處理數百萬的任務。
celery是用Python編寫的，但該協定可以在任何語言實現。它也可以與其他語言透過webhooks實現。
建議的訊息代理RabbitMQ的，但提供有限支援Redis, Beanstalk, MongoDB, CouchDB, ，和資料庫（使用SQLAlchemy的或Django的 ORM） 。
celery是易於整合Django, Pylons and Flask，使用 django-celery, celery-pylons and Flask-Celery 附加套件即可。

2. 安裝
有了上面的概念，需要安裝這麼多東西：RabbitMQ、SQLAlchemy、Celery
安裝方式也都很簡單： RabbitMQ:
mac下：

brew install rabbitmq

linux:

sudo apt-get install rabbitmq-server

##剩下兩個都是Python的東西了，直接pip安裝就好了，對於從來沒有安裝過MySQL驅動的同學可能需要安裝MySQL-python。

安裝完成之後，啟動服務:

$ rabbitmq-server[回车]

#啟動後不要關閉視窗, 下方操作新視窗(Tab)

3. 簡單案例
確保你之前的RabbitMQ已經啟動。 還是官網的那個例子，在任意目錄新建一個tasks.py的文件，內容如下：

#

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

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

在同級目錄執行:

$ celery -A tasks worker --loglevel=info

該指令的意思是啟動一個worker，把tasks中的任務（add(x,y)）把任務放到佇列中。

保持視窗打開，新開一個視窗進入互動模式，python或ipython:

#

>>> from tasks import add
>>> add.delay(4, 4)

##到此為止，你已經可以使用celery執行任務了，上面的python互動模式下簡單的呼叫了add任務，並傳遞4，4參數。

但此時有一個問題，你突然想知道這個任務的執行結果和狀態，到底完了沒有。因此就需要設定backend了。

修改之前的tasks.py中的程式碼為：

# coding:utf-8
import subprocess
from time import sleep

from celery import Celery

backend = 'db+mysql://root:@192.168.0.102/celery'
broker = 'amqp://guest@192.168.0.102:5672'

app = Celery('tasks', backend=backend, broker=broker)


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


@app.task
def hostname():
  return subprocess.check_output(['hostname'])

除了加入backend之外，上面還加入了一個who的方法用來測試多伺服器操作。修改完成之後，還是按照之前的方式啟動。

同樣進入python的互動模型:

>>> from tasks import add, hostname
>>> r = add.delay(4, 4)
>>> r.ready() # 10s内执行，会输出False，因为add中sleep了10s
>>>
>>> r = hostname.delay()
>>> r.result # 输出你的hostname

#4.測試多伺服器

做完上面的測試之後，產生了一個疑惑，Celery叫做分散式任務管理，那它的分散式體現在哪裡？它的任務都是怎麼執行的？在哪個機器上執行的？
在目前伺服器上的celery服務不關閉的情況下，按照相同的方式在另外一台伺服器上安裝Celery，並啟動:

$ celery -A tasks worker --loglevel=info

發現前一個伺服器的Celery服務中輸出你剛啟動的伺服器的hostname，前提是那台伺服器連上了你的rabbitmq。

然後再進入python互動模式:

>>> from tasks import hostname
>>>
>>> for i in range(10):
...   r = hostname.delay()
...   print r.result # 输出你的hostname
>>>

#看你輸入的內容已經觀察兩台伺服器上你啟動celery服務的輸出。

5. RabbitMQ遠端連線的問題

一開始測試時遠端伺服器無法連接本地的RabbitMQ服務，後來發現需要設定權限，在/usr/local/etc/rabbitmq /rabbitmq-env.conf這個檔案中，修改NODE_IP_ADDRESS=127.0.0.1中的ip為0.0.0.0。

6. 總結的說

這篇文章簡單的介紹了Celery的使用，重點還是在分散式的使用。覺得不太爽的地方是，在擴充時，需要重新把程式碼（tasks.py）部署一遍，而不是可以直接把tasks進行共享，可能Celery是透過task來進行不同的worker的匹配的？目前還不太了解，等深入使用後再說。

更多Python環境下安裝使用非同步任務佇列套件Celery相關文章請關注PHP中文網！