FastAPI を使用した Python 非同期 IO をマスターする

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 多重化を簡単に理解するには: 私は宅配便ステーションの上司です。各配達員にタスクの完了について積極的に尋ねる必要はありません。その代わり、配達員は仕事を終えた後、自分で私のところに来てくれます。これにより、タスク処理能力が向上し、より多くのことができるようになります。

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 イベントが完了すると、対応するコールバックがトリガーされ、コルーチンが実行を継続できるようになります。

---

Leapcell: FastAPI およびその他の Python アプリケーションに最適なプラットフォーム:

最後に、Flask/FastAPI をデプロイするための理想的なプラットフォームである Leapcell を紹介します。

Leapcell は、最新の分散アプリケーション向けに特別に設計されたクラウド コンピューティング プラットフォームです。従量課金制の料金モデルにより、アイドルコストは発生しません。つまり、ユーザーは実際に使用したリソースに対してのみ料金を支払います。

WSGI/ASGI アプリケーションに対する Leapcell の独自の利点:

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 中国語 Web サイトの他の関連記事を参照してください。