FastAPI を使用した Python 非同期 IO をマスターする-Python チュートリアル-php.cn

Mastering Python Async IO with FastAPI

Python はインタープリタ型言語であるため、Python Django と組み合わせてバックエンド開発に使用する場合、Java Spring に比べて応答時間が少し長くなります。ただし、コードが合理的である限り、違いはそれほど大きくありません。 Django がマルチプロセスモードを使用している場合でも、その同時処理能力は依然としてはるかに弱いです。 Python には、同時処理能力を向上させるためのソリューションがいくつかあります。たとえば、非同期フレームワーク FastAPI とその非同期機能を使用すると、I/O 集中型タスクの同時処理能力を大幅に強化できます。 FastAPI は、最も高速な Python フレームワークの 1 つです。

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 です。 FastAPI を使用せずに、非同期 I/O を使用して直接非同期処理機能を備えたサーバーを起動できます。

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 操作を待機しないように実行スレッドに指示します。通常の関数の呼び出しはスタックを通じて実装され、関数は 1 つずつしか呼び出して実行できません。ただし、コルーチンは特殊な種類の関数です (共同スレッドではありません)。これにより、スレッドが待機マークで実行を一時停止し、他のタスクの実行に切り替えることができます。 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

ログイン後にコピー

スレッドが 3 つのタスクを 1 つずつ実行しているわけではないことがわかります。 I/O 操作が発生すると、他のタスクの実行に切り替えます。 I/O 操作は完了後も実行を続けます。また、3 つのコルーチンは基本的に同時に 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 await が使用されます。

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 で、他の位置の数値はその上の 2 つの数値の合計です)。

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 や Paul と比較すると、epoll の方がパフォーマンスが優れています。 Linux は通常、デフォルトで epoll を使用し、macOS は epoll に似ており、同様のパフォーマンスを持つ kqueue を使用します。

イベントループを使用したソケットサーバー

pip install fastapi

ログイン後にコピー

指定されたポートを監視するためにサーバーソケットを起動します。 Linux システムで実行している場合、セレクターはデフォルトで実装として epoll を使用します。コードではepollを利用してリクエスト受信イベント(acceptイベント)を登録しています。新しいリクエストが到着すると、epoll はイベント処理関数をトリガーして実行すると同時に、リクエストデータを処理して応答するための読み取りイベント (読み取りイベント) を登録します。 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 イベントが完了すると、対応するコールバックがトリガーされ、コルーチンが実行を継続できるようになります。