Celery 分布式任务队列框架，主要包括三个组成成分：

[1] - Celery 客户端（Client）

[2] - 消息中间件（Message Broker）

[3] - Celery Worker

他们之间的关系如图：

这里，采用 FastAPI 作为 Celery Client，RabbitMQ 作为 Message Broker.

其中，

[1] - Celery Client 运行 FastAPI app，并传递消息或后台任务（message/background jobs） 到 RabbitMQ；

[2] - RabbitMQ 作为 Message Broker，将会调度 clients 和 workers 之间的消息；

[3] - RabbitMQ 在接收到 client 发送的消息后，通过将消息发送到一个 Celery Worker 以初始化 client 任务；

[4] - 一个 Celery Worker 被看做为后台任务，其可以从任何网络服务请求，实现异步性；

[5] - 同时可以有很多 workers 进行或完成很多任务（每个任务作为一个独立线程thread）；

[6] - Celery 确保每个 worker 在同一时间点只执行一个任务，且每个任务只能被分配到一个 worker.

基于以上，实现如下：

Github - azzan-amin-97/FastAPI_Async_Celery

1. 依赖项安装

pip install fastapi
pip install celery
pip install uvicorn #ASGI server to run FastAPI app. 
pip install flower  #任务队列监控

设置 Message Broker（基于 Docker）：

docker run -d --name some-rabbit -p 4369:4369 -p 5671:5671 -p 5672:5672 -p 15672:15672 rabbitmq:3

2. 创建 Celery Worker Task

celery_worker.py

#!/usr/bin/python3
#!--*-- coding: utf-8 --*--
import time

from celery import Celery
from celery.utils.log import get_task_logger

#实例化 Celery
celery = Celery('tasks', broker='amqp://guest:guest@127.0.0.1:5672//')

# 创建 logger，以显示日志信息
celery_log = get_task_logger(__name__)

# 创建任务函数，以订单(Order) 为例，异步进行
@celery.task
def create_order(name, quantity):
    # 5 seconds per 1 order
    complete_time_per_item = 5
    
    # Keep increasing depending on item quantity being ordered
    time.sleep(complete_time_per_item * quantity)
    
    # 显示日志    
    celery_log.info(f"Order Complete!")
    return {"message": f"Hi {name}, Your order has completed!",
            "order_quantity": quantity}

3. 创建 Model 和 App

model.py:

#!/usr/bin/python3
#!--*-- coding: utf-8 --*--
from pydantic import BaseModel

# Pydantic BaseModel
# Order class model for request body
class Order(BaseModel):
    customer_name: str
    order_quantity: int

main.py

#!/usr/bin/python3
#!--*-- coding: utf-8 --*--
from fastapi import FastAPI
from celery_worker import create_order
from model import Order

# Create FastAPI app
app = FastAPI()

# Create order endpoint
@app.post('/order')
def add_order(order: Order):
    # use delay() method to call the celery task
    create_order.delay(order.customer_name, order.order_quantity)
    return {"message": "Order Received! Thank you for your patience."}

4. 运行 FastAPI 和 Celery Worker server

uvicorn main:app --reload

访问：http://localhost:8000/docs 即可查看 FastAPI Swagger Docs.

启动 Celery Worker：

celery -A celery_worker.celery worker --loglevel=info

启动 flower 服务监控 Celery 消息队列：

celery flower -A celery_worker.celery --broker:amqp://localhost//

访问 http://localhost:5555/

5. 测试和分析 Celery

访问 http://localhost:8000/docs，示例如，

点击 Execute，返回结果如：