celery分散式框架實踐
celery簡介
Celery 是一個強大的 分散式任務佇列 的 非同步處理框架，它可以讓任務的執行完全脫離主程式，甚至可以被分配到其他主機上執行。我們通常使用它來實現非同步任務（async task）和定時任務（crontab）。Celery的架構由三部分組成，訊息中介軟體（message broker），任務執行單元（worker）和任務執行結果儲存（task result store）組成。Worker是Celery提供的任務執行的單元，worker併發的執行在分散式的系統節點中。Task result store用來儲存Worker執行的任務的結果，Celery支援以不同方式儲存任務的結果，包括AMQP, redis，memcached, mongodb，SQLAlchemy, Django ORM，Apache Cassandra, IronCache 等。本例中broker和backend都選擇redis。celery的架構如下：
celery和redis安裝
celery安裝
pip install -U Celery
redis安裝
$ wget http://download.redis.io/releases/redis-2.8.17.tar.gz$ tar xzf redis-2.8.17.tar.gz$ cd redis-2.8.17$ make
make完後 redis-2.8.17目錄下會出現編譯後的redis服務程式redis-server,還有用於測試的客戶端程式redis-cli,兩個程式位於安裝目錄 src 目錄下：
下面啟動redis服務.
$ cd src$ ./redis-server
注意這種方式啟動redis 使用的是預設配置。也可以通過啟動引數告訴redis使用指定配置檔案使用下面命令啟動。
$ cd src$ ./redis-server ../redis.conf
redis.conf 是一個預設的配置檔案。我們可以根據需要使用自己的配置檔案。
啟動redis服務程序後，就可以使用測試客戶端程式redis-cli和redis服務互動了。 比如：
$ cd src$ ./redis-cliredis> set foo barOKredis> get foo"bar"
最後安裝python中的redis庫，否則celery提示會報錯
pip install rediscelery實現簡單計算
啟動redis server服務在redis/src目錄下按以上步驟啟動即可編寫任務程式碼tasks.pybackend和broker都用redis
from celery import Celery# 我們這裡案例使用redis作為broker和backendapp = Celery('demo', backend='redis://127.0.0.1:6379/2', broker='redis://127.0.0.1:6379/1')# 建立任務函式@app.taskdef my_task(a, b): print("任務函式正在執行....") return a + b
注意：這裡沒有使用redis密碼，因為redis server預設啟動是不帶密碼的，否則會出現Client sent AUTH, but no password is set的錯誤將tasks模組加入worker
celery -A tasks worker --loglevel=info
可以看到列印以下資訊，任務成功加入
- ** ---------- .> transport: redis://127.0.0.1:6379/1- ** ---------- .> results: redis://127.0.0.1:6379/2- *** --- * --- .> concurrency: 8 (prefork)-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)--- ***** ----- -------------- [queues] .> celery exchange=celery(direct) key=celery [tasks] . tasks.my_task[2019-11-19 23:35:35,537: INFO/MainProcess] Connected to redis://127.0.0.1:6379/1[2019-11-19 23:35:35,544: INFO/MainProcess] mingle: searching for neighbors[2019-11-19 23:35:36,559: INFO/MainProcess] mingle: all alone[2019-11-19 23:35:36,574: INFO/MainProcess] celery@huzhenghui-desktop ready.
執行任務，並將任務結果寫入redis中：
>>> from tasks import my_task>>> ret = my_task.delay(1,2)>>> ret.result3>>>