超實用 Demo：使用 FastAPI、Celery、RabbitMQ 和 MongoDB 實現一個異步任務工作流

作者：somenzz 2022-10-08 07:55:33

數據庫其他數據庫

今天分享一份代碼，使用 Celery、RabbitMQ 和 MongoDB 實現一個異步任務工作流，你可以修改 task.py 來實現你自己的異步任務。

異步任務，是 Web 開發中經常遇到的問題，比如說用戶提交了一個請求，雖然這個請求對應的任務非常耗時，但是不能讓用戶等在這里，通常需要立即返回結果，告訴用戶任務已提交。任務可以在后續慢慢完成，完成后再給用戶發一個完成的通知。

今天分享一份代碼，使用 Celery、RabbitMQ 和 MongoDB 實現一個異步任務工作流，你可以修改 task.py 來實現你自己的異步任務。

架構圖如下：

其中 Celery 來執行異步任務，RabbitMQ 作為消息隊列，MongoDB 存儲任務執行結果，FastAPI 提供 Web 接口。

以上所有模塊均可使用 Docker 一鍵部署。

下面為 Demo 使用方法：

1、確保本機已安裝 Docker、Git

2、下載源代碼：

git clone https://github.com/aarunjith/async-demo.git

3、部署并啟動：

cd async-demo
docker compose up --build

4、啟動一個異步任務：

$ curl -X POST http://localhost:8080/process

任務會發送到消息隊列，同時會立即返回一個任務 id：

? curl -X POST http://localhost:8080/process
{"status":"PENDING","id":"a129c666-7b5b-45f7-ba54-9d7b96a1fe58","error":""}%

5、查詢任務狀態：

curl -X POST http://localhost:8080/check_progress/<task_id>

任務完成后的返回結果如下：

 ? curl -X POST http://localhost:8080/check_progress/a129c666-7b5b-45f7-ba54-9d7b96a1fe58
{"status":"SUCEESS","data":"\"hello\""}%

代碼目錄結構如下：

其中 app.py 如下:

from fastapi import FastAPI
from celery.result import AsyncResult
from tasks import start_processing
from loguru import logger
from pymongo import MongoClient
import uvicorn

# Lets create a connection to our backend where celery stores the results
client = MongoClient("mongodb://mongodb:27017")

# Default database and collection names that Celery create
db = client['task_results']
coll = db["celery_taskmeta"]

app = FastAPI()


@app.post('/process')
async def process_text_file():
    '''
    Process endpoint to trigger the start of a process
    '''
    try:
        result = start_processing.delay()
        logger.info(f'Started processing the task with id {result.id}')
        return {
            "status": result.state,
            'id': result.id,
            'error': ''
        }
    except Exception as e:
        logger.info(f'Task Execution failed: {e}')
        return {
            "status": "FAILURE",
            'id': None,
            'error': e
        }


@app.post('/check_progress/{task_id}')
async def check_async_progress(task_id: str):
    '''
    Endpoint to check the task progress and fetch the results if the task is
    complete.
    '''
    try:
        result = AsyncResult(task_id)
        if result.ready():
            data = coll.find({'_id': task_id})[0]
            return {'status': 'SUCEESS', 'data': data['result']}
        else:
            return {"status": result.state, "error": ''}
    except Exception as e:
        data = coll.find({'_id': task_id})[0]
        if data:
            return {'status': 'SUCEESS', 'data': data['result']}
        return {'status': 'Task ID invalid', 'error': e}

if __name__ == "__main__":
    uvicorn.run("app:app", host='0.0.0.0', port='8080')

如果要實現自己的任務隊列，就修改 task.py 來添加自己的異步任務，可以整合到自己的項目中。

責任編輯：武曉燕來源： Python七號