首页 > 后端开发 > Python教程 > 使用 FastAPI 掌握 Python 异步 IO

使用 FastAPI 掌握 Python 异步 IO

Barbara Streisand
发布: 2025-01-04 19:07:41
原创
619 人浏览过

Mastering Python Async IO with FastAPI

由于Python是解释型语言,当用于后端开发时,例如与Python Django结合时,相对于Java Spring,其响应时间会长一些。不过,只要代码合理,差别并不会太大。即使Django使用多进程模式,其并发处理能力仍然弱很多。 Python有一些提高并发处理能力的解决方案。例如,使用异步框架FastAPI,凭借其异步能力,可以大大增强I/O密集型任务的并发处理能力。 FastAPI 是最快的 Python 框架之一。

FastAPI 为例

我们先简单了解一下如何使用FastAPI。

示例1:默认网络异步IO

安装

pip install fastapi
登录后复制
登录后复制
登录后复制

简单的服务器端代码:

# app.py
from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}
登录后复制
登录后复制
登录后复制

启动

uvicorn app:app --reload
登录后复制
登录后复制
登录后复制

我们可以看到,与其他框架相比,FastAPI的接口只多了一个async关键字。 async 关键字将接口定义为异步。仅从返回结果来看,我们无法看出FastAPI与其他Python框架的区别。区别在于并发访问。 FastAPI的服务器线程在处理路由请求时,如http://127.0.0.1:8000/,如果遇到网络I/O,将不再等待,而是处理其他请求。当网络 I/O 完成时,执行将恢复。这种异步能力提高了 I/O 密集型任务的处理能力。

示例2:显式网络异步IO

让我们看另一个例子。在业务代码中,发起显式的异步网络请求。对于这个网络I/O,就像路由请求一样,FastAPI也会异步处理。

# app.py
from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

# Example of an asynchronous GET request
@app.get("/external-api")
async def call_external_api():
    url = "https://leapcell.io"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code!= 200:
            raise HTTPException(status_code=response.status_code, detail="Failed to fetch data")
        return response.json()
登录后复制
登录后复制
登录后复制

如果希望数据库I/O是异步的,需要数据库驱动或者ORM异步操作的支持。

异步IO

FastAPI异步的核心实现是异步I/O。我们可以直接使用异步I/O来启动一个具有异步处理能力的服务器,而不需要使用FastAPI。

import asyncio

from aiohttp import web

async def index(request):
    await asyncio.sleep(1)  # Simulate I/O operation
    return web.Response(text='{"Hello": "World"}', content_type='application/json')

async def init(loop):
    # Use the event loop to monitor web requests
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    # Start the server, and the event loop monitors and processes web requests
    srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
    print('Server started at http://127.0.0.1:8000...')
    return srv

# Explicitly get an event loop
loop = asyncio.get_event_loop()
# Start the event loop
loop.run_until_complete(init(loop))
loop.run_forever()
登录后复制
登录后复制

本示例启动时,http://127.0.0.1:8000/的返回结果与示例1相同。异步I/O的底层实现原理是“协程”和“事件循环” .

协程

pip install fastapi
登录后复制
登录后复制
登录后复制

函数索引是用 async def 定义的,这意味着它是一个协程。 await 关键字用在 I/O 操作之前,告诉执行线程不要等待本次 I/O 操作。普通函数的调用是通过栈来实现的,函数只能一个一个地调用和执行。然而,协程是一种特殊的函数(不是协作线程)。它允许线程在等待标记处暂停执行并切换到执行其他任务。当I/O操作完成后,会继续执行。

我们来看看多个协程并发执行的效果。

# app.py
from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}
登录后复制
登录后复制
登录后复制

输出:

uvicorn app:app --reload
登录后复制
登录后复制
登录后复制

我们可以看到线程并没有一一执行这三个任务。当它遇到I/O操作时,它会切换到执行其他任务。 I/O操作完成后继续执行。还可以看出,三个协程基本上同时开始等待I/O操作,所以最终的执行完成时间基本相同。虽然这里没有显式使用事件循环,但 asyncio.run 会隐式使用它。

发电机

协程是通过生成器实现的。生成器可以暂停函数的执行,也可以恢复函数的执行,这是协程的特点。

# app.py
from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

# Example of an asynchronous GET request
@app.get("/external-api")
async def call_external_api():
    url = "https://leapcell.io"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code!= 200:
            raise HTTPException(status_code=response.status_code, detail="Failed to fetch data")
        return response.json()
登录后复制
登录后复制
登录后复制

用next()运行生成器时,遇到yield时会暂停。当 next() 再次运行时,它将从上次暂停的地方继续运行。在Python 3.5之前,协程也是用“注释”编写的。从Python 3.5开始,使用async def wait。

import asyncio

from aiohttp import web

async def index(request):
    await asyncio.sleep(1)  # Simulate I/O operation
    return web.Response(text='{"Hello": "World"}', content_type='application/json')

async def init(loop):
    # Use the event loop to monitor web requests
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    # Start the server, and the event loop monitors and processes web requests
    srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
    print('Server started at http://127.0.0.1:8000...')
    return srv

# Explicitly get an event loop
loop = asyncio.get_event_loop()
# Start the event loop
loop.run_until_complete(init(loop))
loop.run_forever()
登录后复制
登录后复制

生成器的暂停和恢复功能除了协程之外还可以用于许多事情。例如,它可以循环计算和存储算法。例如,实现一个帕斯卡三角形(每行两端都是1,其他位置的数字是它上面两个数字的和)。

async def index(request):
    await asyncio.sleep(1)  # Simulate I/O operation
    return web.Response(text='{"Hello": "World"}', content_type='application/json')
登录后复制

输出:

import asyncio
from datetime import datetime

async def coroutine3():
    print(f"Coroutine 3 started at {datetime.now()}")
    await asyncio.sleep(1)  # Simulate I/O operation
    print(f"Coroutine 3 finished at {datetime.now()}")

async def coroutine2():
    print(f"Coroutine 2 started at {datetime.now()}")
    await asyncio.sleep(1)  # Simulate I/O operation
    print(f"Coroutine 2 finished at {datetime.now()}")

async def coroutine1():
    print(f"Coroutine 1 started at {datetime.now()}")
    await asyncio.sleep(1)  # Simulate I/O operation
    print(f"Coroutine 1 finished at {datetime.now()}")

async def main():
    print("Main started")

    # Create tasks to make coroutines execute concurrently
    task1 = asyncio.create_task(coroutine1())
    task2 = asyncio.create_task(coroutine2())
    task3 = asyncio.create_task(coroutine3())

    # Wait for all tasks to complete
    await task1
    await task2
    await task3

    print("Main finished")

# Run the main coroutine
asyncio.run(main())
登录后复制

事件循环

既然协程执行可以暂停,那么协程什么时候恢复执行呢?这就需要使用事件循环来告诉执行线程。

Main started
Coroutine 1 started at 2024-12-27 12:28:01.661251
Coroutine 2 started at 2024-12-27 12:28:01.661276
Coroutine 3 started at 2024-12-27 12:28:01.665012
Coroutine 1 finished at 2024-12-27 12:28:02.665125
Coroutine 2 finished at 2024-12-27 12:28:02.665120
Coroutine 3 finished at 2024-12-27 12:28:02.665120
Main finished
登录后复制

事件循环使用I/O复用技术,不断循环监听协程可以继续执行的事件。当它们可以执行时,线程将继续执行协程。

I/O复用技术

简单理解I/O复用:我是一个快递站的老板。我不需要主动询问每个快递员的任务完成情况。相反,快递员完成任务后会自行来找我。这提高了我的任务处理能力,我可以做更多的事情。

Mastering Python Async IO with FastAPI

select、poll、epoll都可以实现I/O复用。与select和poll相比,epoll具有更好的性能。 Linux一般默认使用epoll,macOS使用kqueue,与epoll类似,性能也差不多。

使用事件循环的套接字服务器

pip install fastapi
登录后复制
登录后复制
登录后复制

启动服务器socket来监听指定端口。如果运行在Linux系统上,选择器默认使用epoll作为其实现。代码中使用epoll来注册一个请求接收事件(accept事件)。当新的请求到来时,epoll会触发并执行事件处理函数,同时注册一个读事件(read event)来处理和响应请求数据。从Web端通过http://127.0.0.1:8000/访问,返回结果与示例1相同。服务器运行日志:

# app.py
from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}
登录后复制
登录后复制
登录后复制

套接字服务器

直接使用Socket启动服务器。使用浏览器访问http://127.0.0.1:8080/或者使用curl http://127.0.0.1:8080/访问时,会返回{"Hello": "World"}

uvicorn app:app --reload
登录后复制
登录后复制
登录后复制

使用curl http://127.0.0.1:8001/访问时,服务器运行日志:

# app.py
from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

# Example of an asynchronous GET request
@app.get("/external-api")
async def call_external_api():
    url = "https://leapcell.io"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code!= 200:
            raise HTTPException(status_code=response.status_code, detail="Failed to fetch data")
        return response.json()
登录后复制
登录后复制
登录后复制

概括

异步I/O在底层使用“协程”和“事件循环”实现。 “协程”确保当线程在执行过程中遇到标记的 I/O 操作时,不必等待 I/O 完成而是可以暂停并让线程执行其他任务而不会阻塞。 “事件循环”使用I/O复用技术,不断循环监视I/O事件。当某个I/O事件完成时,会触发相应的回调,让协程继续执行。


Leapcell:FastAPI 和其他 Python 应用程序的理想平台:

最后介绍一下部署Flask/FastAPI的理想平台:Leapcell。

Leapcell是专为现代分布式应用程序设计的云计算平台。其按需付费的定价模式确保没有闲置成本,这意味着用户只需为他们实际使用的资源付费。

Mastering Python Async IO with FastAPI

Leapcell对于WSGI/ASGI应用的独特优势:

1. 多语言支持

  • 支持 JavaScript、Python、Go 或 Rust 开发。

2. 无限项目免费部署

  • 仅根据使用情况收费。没有要求时不收费。

3. 无与伦比的成本效益

  • 即用即付,无闲置费用。
  • 例如,25 美元可以支持 694 万个请求,平均响应时间为 60 毫秒。

4. 简化的开发者体验

  • 直观的用户界面,易于设置。
  • 完全自动化的 CI/CD 管道和 GitOps 集成。
  • 实时指标和日志,提供可操作的见解。

5. 轻松的可扩展性和高性能

  • 自动伸缩,轻松应对高并发。
  • 零运营开销,让开发者专注于开发。

在文档中了解更多信息!

Leapcell Twitter:https://x.com/LeapcellHQ

以上是使用 FastAPI 掌握 Python 异步 IO的详细内容。更多信息请关注PHP中文网其他相关文章!

来源:dev.to
本站声明
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn
作者最新文章
热门教程
更多>
最新下载
更多>
网站特效
网站源码
网站素材
前端模板