基於Celery在多臺雲服務器上實現分佈式

起源

最近參加公司裏的一個比賽，比賽內容裏有一項是儘可能使用分佈式實現項目。由於項目最終會跑在jetsonnano，一個賊卡的開發板，性能及其垃圾。並且要求使用python？
找了不少博客，講的真的是模棱兩可，最後結合官方文檔終於啃出來，寫出來分享一下。

目前本博客的幾個要點：

1. 使用celery實現分佈式
2. 結合flask框架使用
3. 使用Redis做爲broker
4. 使用Redis做爲backend

準備工做

安裝環境

須要安裝的環境包括Redis、Celery、Pyhton3.6.9「開發板自帶」

這裏須要填坑的是Redis的設置。由於在騰訊雲服務器、華爲雲服務器、滴滴雲服務器上作的測試「窮，每一個平臺褥點羊毛」，一開始不知道在安全組裏面開放端口，一直鏈接不上服務器，很坑。這裏若是沒有使用雲服務器的話能夠跳過安全組這一步的設置，使用了雲服務器的話，一件開放一下端口，自行百度開放方法。

首先是Redis的設置，Redis須要把默認的120.0.0.1 IP地址修改成0.0.0.0，並把守護進程關閉。

vim redis.conf
bind = 0.0.0.0
protected -mode no //從yes改成no

安裝celery

pip3 install celery -i https://pypi.tuna.tsinghua.edu.cn/simple

一個簡單的分佈式Demo

網上關於建立Celery文件的描述都很模糊，這裏個人理解是這樣，首先看一下官方給出的Demo：

from celery import Celery

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

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

將這個文件命名爲tasks.py

這段代碼最關鍵的核心是app = Celery('tasks', broker='pyamqp://guest@localhost//')
既然是分佈式，確定要有worker，幹活的人，也就是雲服務器，在雲服務器上須要作的事情是：celery -A tasks worker --loglevel=info
其中的celery -A *** worker --loglevel=info *** 就是worker要拿到任務的任務板標誌，只有有個這個標誌，worker才知道到底誰在發任務。

broker='pyamqp://guest@localhost//' 由於我使用redis來做爲任務的存放容器，因此改成 broker='redis://guest@localhost//'
broker是存聽任務的地方，因此我把發聽任務的服務器的地址填進去： app = Celery('tasks', broker='redis://121.***.***.190:6379')
6379爲默認的端口號，其實broker這段url應該包括redis的用戶名、用戶密碼+IP地址。由於咱們前面修改的redis的配置文件，因此這裏能夠無密碼訪問。

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

這一段就是服務器要發送出去的任務了。固然服務器裏不須要包含執行任務所須要的庫，庫安裝在worker的服務器裏就能夠了。「固然add(x,y)啥庫也不須要」。

如今能夠來見識一下celery的威力了，把上面修改後的tasks.py放到worker服務器上面，執行命令：celery -A tasks worker --loglevel=info
你會看到下面這行：

(base) zhaosi@zhaosideMBP *** % celery -A tasks worker --loglevel=info
 
 -------------- celery@zhaosideMBP v4.4.7 (cliffs)
--- ***** ----- 
-- ******* ---- macOS-10.15.6-x86_64-64bit 2020-09-05 14:35:13
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7fd0bb1b16a0
- ** ---------- .> transport:   redis://121.***.***.190:6379/8
- ** ---------- .> results:     redis://121.***.***.190:6379/7
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                
[tasks]
  . tasks.add

[2020-09-05 14:35:15,566: INFO/MainProcess] Connected to redis://121.***.***.190:6379/8
[2020-09-05 14:35:15,773: INFO/MainProcess] mingle: searching for neighbors
[2020-09-05 14:35:17,484: INFO/MainProcess] mingle: all alone
[2020-09-05 14:35:18,789: INFO/MainProcess] celery@zhaosideMBP ready.

當你看到最後四行時，YES，最簡單的Demo被你跑起來了

[tasks]
  . tasks.add

這裏展現的是worker能夠接到的任務，固然如今服務器尚未發佈任務，worker在持續監聽服務器上存儲發佈任務的redis數據庫，等着接活。

驗收成果

打開雲服務器，準備發佈任務「請在服務器上也建立一個tasks.py,不須要安裝任何依賴」

root@-x:~/pro# python3
Python 3.6.9 (default, Jul 17 2020, 12:50:27) 
[GCC 8.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from tasks import add
>>> add.delay(1,1)
<AsyncResult: 71c49a0e-2d52-444e-b158-1ae9f1486767>
>>>

回到worker服務器，能夠看到任務被接收而且完成了！

[2020-09-05 14:41:10,138: INFO/MainProcess] Received task: tasks.add[71c49a0e-2d52-444e-b158-1ae9f1486767]  
[2020-09-05 14:41:10,223: WARNING/ForkPoolWorker-8] 2

能夠經過result來接受結果：

>>> result = add.delay(1,1)
<AsyncResult: 71c49a0e-2d52-444e-b158-1ae9f1486767>
>>> result.get()
2

更復雜的返回值請各位自行探索啦
所謂萬事開頭難，有了這個Demo的幫助，後續的任務會簡單不少。

後續

仍是官方文檔好啊。如今大多數博客寫的是個啥。