如何快速創建一個擁有異步任務隊列集群的 Rest Api
異步任務是 Web 后端開發中最常見的需求，非常適合多任務、高并發的場景。本文分享如何使用 docker-compose、FastAPI、rq 來快速創建一個包含異步任務隊列集群的 REST API，后端執行任務的節點可以隨意擴展。
系統的架構圖：
上圖中的每一個方框都可以理解為一個服務器。
用戶請求 api， api 將任務放入 redis 隊列，worker 自動去 redis 隊列取出任務并執行，worker 節點可以任意水平擴展。
接下來，我們來實現這一架構的 demo，你可以看到 docker 的強大和方便之處。
1、先創建一個虛擬環境，安裝依賴
依賴 fastapi，redis，rq 庫，安裝后生成一個 requirements.txt 文件
- mkdir myproject
- python3 -m venv env
- source env/bin/activate
- pip install rq
- pip install fastapi
- pip install redis
- pip freeze > requirements.txt
2、編碼實現 REST API、Worker
REST 是一種風格，這里不是重點，我們使用 FastAPI 來快速創建一個接口，新建一個 api.py 的文件，內容如下：
- from fastapi import FastAPI
- from redis import Redis
- from rq import Queue
- from worker import send_captcha
- app = FastAPI()
- # 需要注意，這里的 host 是主機名，在 docker 中就是服務名，后面的 docker-compose.ymal 中的服務名稱也要是這個
- redis_conn = Redis(host='myproj_redis', port=6379, db=0)
- # 定義一個隊列，名稱是 my_queue
- q = Queue('my_queue', connection=redis_conn)
- @app.get('/hello')
- def hello():
- """Test endpoint"""
- return {'hello': 'world'}
- # Rest API 示例
- @app.post('/send_captcha/{phone_number}', status_code=201)
- def addTask(phone_number: str):
- """
- Adds tasks to worker queue.
- Expects body as dictionary matching the Group class.
- """
- job = q.enqueue(send_captcha, phone_number)
- return {'job': "tasks add done."}
這里的 send_captcha 函數就是一個異步任務，從 worker.py 中導入，worker.py 的內容如下：
- import time
- def send_captcha(phone_number):
- """
- 模擬一個耗時的異步任務
- """
- print(f'{time.strftime("%T")} 準備發送手機驗證碼') # in place of actual logging
- print(f'{time.strftime("%T")} 生成隨機驗證碼并存入 redis，設置 5 分鐘過期時間')
- time.sleep(5) # simulate long running task
- print(f'{time.strftime("%T")} {phone_number}發送完成')
- return { phone_number: 'task complete'}
return { phone_number: 'task complete'}
3、構建 Dokcer 鏡像
現在的目標是實現一個擁有兩個執行節點的集群。我們需要啟動 4 個容器來完成一個集群部署：
- 容器1：運行 FastAPI app
- 容器2：運行 Redis 服務
- 容器3：運行 worker 1 服務
- 容器4：運行 worker 2 服務
其中容器 1、3、4 都是 Python 應用，可以共用一個 Python 鏡像。
為了方便調試，我們可以讓 1、3、4 容器共享我們的本地路徑，這樣改了代碼就不需要重新構建鏡像，比較方便。
創建一個包含依賴的 Python 鏡像
現在我們來創建一個包含前文 requirements.txt 依賴的 Python 鏡像，編寫 Dockerfile，內容如下：
- FROM python:3.8-alpine
- RUN adduser -D myproj
- WORKDIR /home/myproj
- COPY requirements.txt requirements.txt
- RUN pip install -r requirements.txt
- RUN chown -R myproj:myproj ./
- USER myproj
- CMD uvicorn api:app --host 0.0.0.0 --port 5057
內容說明：
FROM python:3.8-alpine
指定使用 python:3.8-alpine，這個容器已經預裝了 Python3.8，可以在命令行執行 docker search python 看看有哪些 Python 鏡像。
RUN adduser -D myproj
添加一個用戶 myproj，這一步的主要目的是為了生成目錄 /home/myproj
WORKDIR /home/myproj
設置程序的執行路徑為 /home/myproj
COPY requirements.txt requirements.txt
復制當前路徑下的 requirements.txt 到容器的 /home/myproj，這里沒有復制 .py 文件是因為后面我們啟動容器的時候會共享本地路徑，不需要再復制了，生產部署時最好復制到窗口內部，這樣容器就不會依賴本機。
RUN pip install -r requirements.txt
在容器中安裝依賴
RUN chown -R myproj:myproj ./
將 /home/myproj 路徑下的文件的擁有者和所屬組改為 myproj，這一步為了使用 myproj 用戶來啟動 fastapi 服務，生產環境通常用 root 用戶啟動，也就不需要這個指令了。
USER myproj
切換到 myproj 用戶
CMD uvicorn api:app --host 0.0.0.0 --port 5057
容器啟動后執行的命令，服務端口為 5057
更多的 Dockerfile 語法請參考官方文檔，這里僅是簡要說明。
現在 Dockerfile 所在的目錄執行下面的命令構建一個鏡像：
- docker build -t myproject:latest .
創建完成后，可以使用 docker images 來查看：
- ❯ docker images | grep myproj
- myproject
4、啟動集群
這里使用 Docker Compose 來啟動 4 個容器，為什么用 Docker Compose 呢?因為方便，如果不用的話，需要手動一個容器一個容器啟動。
Docker Compose 會讀取一個 yaml 格式的配置文件，依據配置文件來啟動容器，各容器共享同一網絡。還記得 api.py 中使用的 Redis 主機名嗎，這里就需要將 redis 服務名設置為那個主機名。
編寫一個 docker-compose.yml 內容如下：
- version: '3'
- services:
- myproj_redis:
- image: redis:4.0-alpine
- ports:
- - "6379:6379"
- volumes:
- - ./redis:/data
- myproj_api:
- image: myproject:latest
- command: uvicorn api:app --host 0.0.0.0 --port 5057
- ports:
- - "5057:5057"
- volumes:
- - ./:/home/myproj
- myproj_worker1:
- image: myproject:latest
- command: rq worker --url redis://myproj_redis:6379 my_queue
- volumes:
- - ./:/home/myproj
- myproj_worker2:
- image: myproject:latest
- command: rq worker --url redis://myproj_redis:6379 my_queue
- volumes:
- - ./:/home/myproj
第一個容器是 myproj_redis，運行著 redis 服務， redis 的數據通過 volumes 方式保存在本地，因此需要在本地創建一個 redis 目錄，來映射容器內部的 /data 目錄。
第二個容器就是 fastapi 服務，端口 5057，使用本地路徑映射為 /home/myproj
第三個容器和第四個容器是 worker 節點，雖然也映射了本地路徑，但它僅使用 worker.py 文件。當任務太多時，worker 節點可以擴展，解決負載壓力，
最終的目錄是這樣：
執行 docker compose 命令啟動 4 個容器:
- docker compose -f docker-compose.yml up
可以看到 4 個服務均啟動并正常打印了日志輸出。
4、測試
現在來測試一下，左邊的窗口，我使用 Python 快速發送了 3 個 post 請求:
- import subprocess
- for i in range(3):
- subprocess.run("curl -v -X POST 'http://localhost:5057/send_captcha/18012345678'",shell = True)
從右邊窗口的日志輸出可以看出 worker1 和 worker2 都執行了任務，其中 worker1 執行了 2 個，worker2 執行了 1 個。
查看完整代碼請點擊「閱讀原文」
最后的話
本文分享了如何使用 Dockerfile 構建一個鏡像，使用 Docker Compose 管理一個容器集群，以此為基礎實現了一個具有異步任務隊列集群的 REST API，拋磚引玉，關于 Dockerfile、docker-compose 的詳細用法，還請參考 Docker 官方文檔
