Python の非同期をマスターする: コルーチンとイベントループでアプリのパフォーマンスを向上させる-Python チュートリアル-php.cn

Mastering Python

Python の非同期プログラミングは、高性能アプリケーションの構築に大きな変革をもたらします。私はこれを何年も使っていますが、正しく使えばどれほど強力になるかにいつも驚かされます。

Python の非同期モデルの中心となるのは、コルーチンとイベントループです。コルーチンは、実行を一時停止および再開できる特別な関数であり、スレッドのオーバーヘッドなしで効率的なマルチタスクを可能にします。一方、イベントループは、これらのコルーチンを駆動し、実行を管理し、I/O 操作を処理するエンジンです。

コルーチンから始めましょう。 Python では、async def 構文を使用して定義します。簡単な例を次に示します:

async def greet(name):
    print(f"Hello, {name}!")
    await asyncio.sleep(1)
    print(f"Goodbye, {name}!")

ログイン後にコピー

このコルーチンは人に挨拶し、少し待ってから別れを告げます。ここでは await キーワードが重要です。これにより、コルーチンが実行を一時停止し、制御をイベントループに戻すことができます。

しかし、コルーチンは内部でどのように機能するのでしょうか?これらは実際には、Python のジェネレーター機能の上に構築されています。コルーチンを呼び出しても、すぐには実行されません。代わりに、コルーチンオブジェクトを返します。このオブジェクトは、ジェネレーターと同様に、値を送信したり、値を生成したりできます。

イベントループは、これらのコルーチンを実際に実行する役割を果たします。タスク (コルーチンのラッパー) のキューを維持し、それらを 1 つずつ実行します。コルーチンが await ステートメントに到達すると、イベントループはそれを一時停止し、次のタスクに進みます。これが協調的なマルチタスクの本質です。タスクは自発的に制御を放棄し、他のタスクが実行できるようにします。

イベントループがどのように機能するかを簡略化して示します。

class EventLoop:
    def __init__(self):
        self.ready = deque()
        self.sleeping = []

    def call_soon(self, callback):
        self.ready.append(callback)

    def call_later(self, delay, callback):
        deadline = time.time() + delay
        heapq.heappush(self.sleeping, (deadline, callback))

    def run_forever(self):
        while True:
            self.run_once()

    def run_once(self):
        now = time.time()
        while self.sleeping and self.sleeping[0][0] <= now:
            _, callback = heapq.heappop(self.sleeping)
            self.ready.append(callback)

        if self.ready:
            callback = self.ready.popleft()
            callback()
        else:
            time.sleep(0.1)  # Avoid busy waiting

ログイン後にコピー

このイベントループは、実行準備ができているタスク (準備完了キュー内) とスリープ中のタスク (スリープリスト内) の 2 種類のタスクを管理します。 run_forever メソッドは、タスクがなくなるまでタスクを実行し続けます。

それでは、タスクのスケジュール設定について話しましょう。 Python の asyncio モジュールは、高度なスケジューリング機能を備えた、より洗練されたイベントループを提供します。 I/O 操作を処理し、サブプロセスを実行し、他のイベントループと統合することもできます。

asyncio を使用して複数のコルーチンを同時に実行する方法は次のとおりです。

import asyncio

async def task1():
    print("Task 1 starting")
    await asyncio.sleep(2)
    print("Task 1 finished")

async def task2():
    print("Task 2 starting")
    await asyncio.sleep(1)
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

ログイン後にコピー

このスクリプトは両方のタスクを開始しますが、タスク 2 はスリープ時間が短いため、タスク 1 よりも前に終了します。

非同期プログラミングの最も強力なアプリケーションの 1 つは、ネットワーク操作です。単純な非同期 Web サーバーを見てみましょう:

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    response = f"Echo: {message}\n"
    writer.write(response.encode())
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

ログイン後にコピー

このサーバーはスレッドを使用せずに複数のクライアントを同時に処理できるため、非常に効率的です。

しかし、非同期プログラミングはサーバーだけのものではありません。これは、特に複数のネットワーク要求を行う必要がある場合に、クライアントにとっても最適です。これは、複数のページを同時にフェッチできるシンプルな Web スクレイパーです:

async def greet(name):
    print(f"Hello, {name}!")
    await asyncio.sleep(1)
    print(f"Goodbye, {name}!")

ログイン後にコピー

このスクレイパーは複数のページを同時にフェッチできるため、同期アプローチと比較してプロセスが大幅に高速化されます。

ここで、より高度な概念をいくつか見ていきましょう。 Python の非同期モデルの興味深い機能の 1 つは、独自のイベントループを作成できることです。これは、非同期コードを他のフレームワークと統合する必要がある場合、または特定のユースケースに合わせて最適化したい場合に役立ちます。

これは、同期コールバックと非同期コールバックの両方を実行できる単純なカスタムイベントループです。

class EventLoop:
    def __init__(self):
        self.ready = deque()
        self.sleeping = []

    def call_soon(self, callback):
        self.ready.append(callback)

    def call_later(self, delay, callback):
        deadline = time.time() + delay
        heapq.heappush(self.sleeping, (deadline, callback))

    def run_forever(self):
        while True:
            self.run_once()

    def run_once(self):
        now = time.time()
        while self.sleeping and self.sleeping[0][0] <= now:
            _, callback = heapq.heappop(self.sleeping)
            self.ready.append(callback)

        if self.ready:
            callback = self.ready.popleft()
            callback()
        else:
            time.sleep(0.1)  # Avoid busy waiting

ログイン後にコピー

このカスタムループは非常に基本的なものですが、中心となる原則を示しています。これを拡張して、I/O 操作やタイマーなどのより複雑なシナリオを処理できます。

非同期コードのデバッグは、特に複雑なアプリケーションを扱う場合には困難になることがあります。私が便利だと思うテクニックの 1 つは、asyncio のデバッグモードを使用することです。次のように有効にできます:

import asyncio

async def task1():
    print("Task 1 starting")
    await asyncio.sleep(2)
    print("Task 1 finished")

async def task2():
    print("Task 2 starting")
    await asyncio.sleep(1)
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

ログイン後にコピー

これにより、完了しないコルーチンや実行に時間がかかりすぎるコールバックなどに関する、より詳細なエラーメッセージと警告が表示されます。

もう 1 つの便利なデバッグ手法は、asyncio のタスクイントロスペクション機能を使用することです。たとえば、実行中のすべてのタスクのリストを取得できます:

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    response = f"Echo: {message}\n"
    writer.write(response.encode())
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

ログイン後にコピー

これは、プログラムがその時点で何を行っているかを理解するのに役立ちます。

非同期コードを最適化する場合、重要な原則の 1 つは、同期操作にかかる時間を最小限に抑えることです。長時間実行される同期コードはイベントループをブロックし、他のコルーチンの実行を妨げます。 CPU を集中的に使用するタスクがある場合は、別のスレッドまたはプロセスで実行することを検討してください。

もう 1 つの最適化テクニックは、同時に実行できるコルーチンが複数ある場合に asyncio.gather を使用することです。これは、それらを 1 つずつ待つよりも効率的です:

import asyncio
import aiohttp

async def fetch_page(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'http://example.com',
        'http://example.org',
        'http://example.net'
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_page(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)

    for url, page in zip(urls, pages):
        print(f"Page from {url}: {len(page)} bytes")

asyncio.run(main())

ログイン後にコピー

最後に、非同期プログラミングが常に最良の解決策であるとは限らないことを覚えておいてください。待機時間が長い I/O バウンドのタスクの場合、パフォーマンスが大幅に向上します。ただし、CPU に依存するタスクの場合は、従来のマルチスレッドまたはマルチプロセッシングの方が適切な場合があります。

結論として、コルーチンとイベントループに基づいて構築された Python の非同期プログラミングモデルは、効率的でスケーラブルなアプリケーションを作成するための強力な方法を提供します。 Web サーバー、ネットワーククライアント、データ処理パイプラインのいずれを構築している場合でも、これらの概念を理解すると、Python の非同期機能を最大限に活用することができます。他の強力なツールと同様、効果的に使用するには練習と慎重な思考が必要ですが、結果は本当に印象的です。