Celery （芹菜）是基於Python開發的分散式任務佇列。它支援使用任務佇列的方式在分佈的機器／程序／執行緒上執行任務排程。

架構設計

Celery的架構由三部分組成，訊息中介軟體（message broker），任務執行單元（worker）和任務執行結果儲存（task result store）組成。

• 訊息中介軟體

  Celery本身不提供訊息服務，但是可以方便的和第三方提供的訊息中介軟體整合。包括，RabbitMQ, Redis, MongoDB (experimental), Amazon SQS (experimental),CouchDB (experimental), SQLAlchemy

   (experimental),Django ORM (experimental), IronMQ

• 任務執行單元

  Worker是Celery提供的任務執行的單元，worker併發的執行在分散式的系統節點中。

• 任務結果儲存

  Task result store用來儲存Worker執行的任務的結果，Celery支援以不同方式儲存任務的結果，包括AMQP, Redis，memcached, MongoDB，SQLAlchemy, Django ORM，Apache Cassandra, IronCache

另外， Celery還支援不同的併發和序列化的手段

• 序列化

  pickle, json, yaml

  , msgpack. zlib, bzip2 compression， Cryptographic message signing 等等

安裝和執行

Celery的安裝過程略為複雜，下面的安裝過程是基於我的AWS EC2的Linux版本的安裝過程，不同的系統安裝過程可能會有差異。大家可以參考官方文件。

首先我選擇RabbitMQ作為訊息中介軟體，所以要先安裝RabbitMQ。作為安裝準備，先更新YUM。

sudo yum -y update

RabbitMQ是基於erlang的，所以先安裝erlang

# Add and enable relevant application repositories:
# Note: We are also enabling third party remi package repositories.
wget http://dl.fedoraproject.org/pub/epel/6/x86_64/epel-release-6-8.noarch.rpm
wget http://rpms.famillecollet.com/enterprise/remi-release-6.rpm
sudo rpm -Uvh remi-release-6*.rpm epel-release-6*.rpm

# Finally, download and install Erlang:
yum install -y erlang
 

然後安裝RabbitMQ

# Download the latest RabbitMQ package using wget:
wget  
# Add the necessary keys for verification:
rpm --import  
# Install the .RPM package using YUM:
yum install rabbitmq-server-3.2.2-1.noarch.rpm

啟動RabbitMQ服務

rabbitmq-server start

RabbitMQ服務已經準備好了，然後安裝Celery， 假定你使用pip來管理你的python安裝包

pip install Celery

為了測試Celery是否工作，我們執行一個最簡單的任務，編寫tasks.py

from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://[email protected]//')
app.conf.CELERY_RESULT_BACKEND = 'db+sqlite:///results.sqlite'

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

在當前目錄執行一個worker，用來執行這個加法的task

celery -A tasks worker --loglevel=info

其中－A引數表示的是Celery App的名字。注意這裡我使用的是SQLAlchemy作為結果儲存。對應的python包要事先安裝好。

worker日誌中我們會看到這樣的資訊

- ** ---------- [config]
- ** ---------- .> app:         tasks:0x1e68d50
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     db+sqlite:///results.sqlite
- *** --- * --- .> concurrency: 8 (prefork)

其中，我們可以看到worker預設使用prefork來執行併發，並設定併發數為8

下面的任務執行的客戶端程式碼：

from tasks import add
import time
result = add.delay(4,4)

while not result.ready():
  print "not ready yet"
  time.sleep(5)

print result.get()

用python執行這段客戶端程式碼，在客戶端，結果如下

not ready   
8

Work日誌顯示

[2015-03-12 02:54:07,973: INFO/MainProcess] Received task: tasks.add[34c4210f-1bc5-420f-a421-1500361b914f]
[2015-03-12 02:54:08,006: INFO/MainProcess] Task tasks.add[34c4210f-1bc5-420f-a421-1500361b914f] succeeded in 0.0309705100954s: 8

這裡我們可以發現，每一個task有一個唯一的ID，task非同步執行在worker上。

這裡要注意的是，如果你執行官方文件中的例子，你是無法在客戶端得到結果的，這也是我為什麼要使用SQLAlchemy來儲存任務執行結果的原因。官方的例子使用AMPQ，有可能Worker在列印日誌的時候取出了task的執行結果顯示在worker日誌中，然而AMPQ作為一個訊息佇列，當訊息被取走後，佇列中就沒有了，於是客戶端總是無法得到任務的執行結果。不知道為什麼官方文件對這樣的錯誤視而不見。

如果大家想要對Celery做更進一步的瞭解，請參考官方文件