Python環(huán)境下安裝使用異步任務隊列包Celery的基礎教程

更新時間：2016年05月07日 18:03:42 投稿：goldensun

這篇文章主要介紹了Python環(huán)境下安裝使用異步任務隊列包Celery的基礎教程,Celery的分布式任務管理適合用于服務器集群的管理和維護,需要的朋友可以參考下

1.簡介

celery(芹菜)是一個異步任務隊列/基于分布式消息傳遞的作業(yè)隊列。它側重于實時操作，但對調度支持也很好。
celery用于生產系統(tǒng)每天處理數(shù)以百萬計的任務。
celery是用Python編寫的，但該協(xié)議可以在任何語言實現(xiàn)。它也可以與其他語言通過webhooks實現(xiàn)。
建議的消息代理RabbitMQ的，但提供有限支持Redis, Beanstalk, MongoDB, CouchDB, ，和數(shù)據(jù)庫（使用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已經(jīng)啟動。
還是官網(wǎng)的那個例子，在任意目錄新建一個tasks.py的文件，內容如下：

from celery import Celery

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

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

在同級目錄執(zhí)行:

$ celery -A tasks worker --loglevel=info

該命令的意思是啟動一個worker，把tasks中的任務（add(x,y)）把任務放到隊列中。
保持窗口打開，新開一個窗口進入交互模式，python或者ipython:

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

到此為止，你已經(jīng)可以使用celery執(zhí)行任務了，上面的python交互模式下簡單的調用了add任務，并傳遞4，4參數(shù)。
但此時有一個問題，你突然想知道這個任務的執(zhí)行結果和狀態(tài)，到底完了沒有。因此就需要設置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內執(zhí)行，會輸出False，因為add中sleep了10s
>>>
>>> r = hostname.delay()
>>> r.result # 輸出你的hostname

4. 測試多服務器
做完上面的測試之后，產生了一個疑惑，Celery叫做分布式任務管理，那它的分布式體現(xiàn)在哪？它的任務都是怎么執(zhí)行的？在哪個機器上執(zhí)行的？
在當前服務器上的celery服務不關閉的情況下，按照同樣的方式在另外一臺服務器上安裝Celery，并啟動:

$ celery -A tasks worker --loglevel=info

發(fā)現(xiàn)前一個服務器的Celery服務中輸出你剛啟動的服務器的hostname，前提是那臺服務器連上了你的rabbitmq。
然后再進入python交互模式:

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

看你輸入的內容已經(jīng)觀察兩臺服務器上你啟動celery服務的輸出。

5. RabbitMQ遠程連接的問題
一開始測試時遠程服務器無法連接本地的RabbitMQ服務，后來發(fā)現(xiàn)需要設置權限，在/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的匹配的？目前還不太了解，等深入使用之后再說。

您可能感興趣的文章: