正文

揭秘FastAPI结合Celery实现高效异步任务处理，实战案例解析

/0 浏览量
1013

在当今快速发展的互联网时代，异步任务处理已经成为提高应用性能和响应速度的关键技术。FastAPI和Celery是Python中两个非常流行的框架，分别用于构建高性能的API和异步任务队列。本文将深入探讨如何将FastAPI与Celery结合，实现高效异步任务处理，并通过实战案例进行解析。

FastAPI简介

FastAPI是一个现代、快速（高性能）的Web框架，用于构建API，与Python 3.6+类型提示一起使用。它具有以下特点：

  • 高性能：使用Starlette和Uvicorn，提供快速的响应速度。
  • 易于使用：具有简洁的语法和强大的类型提示。
  • 自动文档：自动生成交互式API文档。
  • 依赖注入：内置依赖注入系统，简化代码编写。

Celery简介

Celery是一个异步任务队列/作业队列基于分布式消息传递的开源项目。它专注于实时处理，同时也支持任务调度。Celery具有以下特点：

  • 分布式：支持在多个机器上运行，实现负载均衡。
  • 可靠：提供丰富的失败重试机制。
  • 灵活：支持多种消息代理，如RabbitMQ、Redis等。
  • 易于扩展：可以轻松地添加新的任务和消息代理。

FastAPI结合Celery实现异步任务处理

将FastAPI与Celery结合，可以实现高效的异步任务处理。以下是一个简单的示例：

1. 安装依赖

首先，需要安装FastAPI、Celery和Uvicorn：

pip install fastapi celery[redis] uvicorn

2. 创建FastAPI应用

创建一个名为main.py的文件，并编写以下代码：

from fastapi import FastAPI
from celery import Celery

app = FastAPI()

# 配置Celery
celery_app = Celery(
    'tasks',
    backend='redis://localhost:6379/0',
    broker='redis://localhost:6379/0',
)

# 定义异步任务
@celery_app.task
def add(x, y):
    return x + y

@app.post("/add/")
async def perform_add(x: int, y: int):
    result = add.delay(x, y)
    return {"result": result.get(timeout=10)}

3. 运行应用

使用Uvicorn运行FastAPI应用：

uvicorn main:app --reload

4. 测试API

访问以下URL进行测试：

POST /add/
{
  "x": 1,
  "y": 2
}

返回结果：

{
  "result": 3
}

实战案例解析

以下是一个使用FastAPI和Celery实现的邮件发送任务的实战案例：

1. 邮件发送任务

首先，需要安装python-multipart和smtplib库：

pip install python-multipart smtplib

然后，在tasks.py文件中定义邮件发送任务：

import smtplib
from email.mime.text import MIMEText
from email.header import Header

@celery_app.task
def send_email(subject, content, to_email):
    msg = MIMEText(content, 'plain', 'utf-8')
    msg['From'] = Header("FastAPI", 'utf-8')
    msg['To'] = Header(to_email, 'utf-8')
    msg['Subject'] = Header(subject, 'utf-8')

    try:
        smtp_obj = smtplib.SMTP('smtp.example.com', 587)
        smtp_obj.starttls()
        smtp_obj.login('username@example.com', 'password')
        smtp_obj.sendmail('username@example.com', to_email, msg.as_string())
        smtp_obj.quit()
        return "邮件发送成功"
    except Exception as e:
        return f"邮件发送失败：{e}"

2. 测试邮件发送任务

在main.py中添加以下代码：

@app.post("/send_email/")
async def send_email(subject: str, content: str, to_email: str):
    result = send_email.delay(subject, content, to_email)
    return {"result": result.get(timeout=10)}

访问以下URL进行测试：

POST /send_email/
{
  "subject": "测试邮件",
  "content": "这是一封测试邮件。",
  "to_email": "recipient@example.com"
}

返回结果：

{
  "result": "邮件发送成功"
}

通过以上实战案例，我们可以看到FastAPI和Celery结合实现异步任务处理的强大能力。在实际项目中，可以根据需求添加更多的任务和功能，以满足不同的业务场景。

-- 展开阅读全文 --