Pythonの非同期プログラミング

Pythonでは、コルーチン、マルチスレッディング、マルチプロセッシングなど、複数の非同期アプローチが利用可能です。さらに、いくつかの従来の方法とサードパーティの非同期ライブラリがあります。この記事では、主にコルーチンに焦点を当て、マルチスレッディングとマルチプロセッシングについて簡単に紹介します。

async/await

Pythonでは、asyncで宣言された関数は非同期関数であり、しばしばコルーチンと呼ばれます。例：

import asyncio

async def hello():
    await asyncio.sleep(1)
    print("hello leapcell")

呼び出し方法

非同期関数の呼び出し方は、通常の関数とは少し異なります。たとえば、通常の関数の呼び出し方は次のとおりです。

def hello():
    print("hello leapcell")

hello()

そして、非同期関数の呼び出し方は次のとおりです。

import asyncio

async def hello():
    await asyncio.sleep(1)
    print("hello leapcell")

h = hello()
asyncio.run(h)

非同期関数を呼び出すとき、最初にh = hello()を使用するとコルーチンオブジェクトが返され、関数内のコードは実行されません。 asyncio.run(h)関数またはawait hステートメントを使用して初めてコードが実行されます。以下に示すように：

import asyncio

async def async_function():
    print("This is inside the async function")
    await asyncio.sleep(1)
    return "Async function result"

# 正しい使い方
async def correct_usage():
    print("Correct usage:")
    result = await async_function()
    print(f"Result: {result}")

# awaitを使用せずに呼び出す
def incorrect_usage():
    print("\nIncorrect usage:")
    coroutine = async_function()
    print(f"Returned object: {coroutine}")
    # 注: "This is inside the async function" はここでは出力されません

# 未awaitのコルーチンを処理する
async def handle_unawaited_coroutine():
    print("\nHandling unawaited coroutine:")
    coroutine = async_function()
    try:
        # asyncio.run()を使用してコルーチンを実行する
        result = await coroutine
        print(f"Result after handling: {result}")
    except RuntimeWarning as e:
        print(f"Caught warning: {e}")

async def main():
    await correct_usage()
    incorrect_usage()
    await handle_unawaited_coroutine()

asyncio.run(main())

非同期関数を呼び出すための一般的な方法

asyncio.gather()

gatherを使用すると、複数のタスクが同時に開始され、並行して実行されます。すべてのタスクが完了して結果が返された後、後続のコードが実行され続けます。例：

import asyncio

async def num01():
    await asyncio.sleep(1)
    return 1

async def num02():
    await asyncio.sleep(1)
    return 2

async def combine():
    results = await asyncio.gather(num01(), num02())
    print(results)

asyncio.run(combine())

出力:

[1, 2]

上記には2つの非同期関数があります。次に、asyncio.gatherを使用して、これらの2つの関数を同時に実行し、awaitを使用して結果を待ちます。返された結果はresultsに保存されます。

awaitを直接使用する

上記のgatherメソッドは、複数の非同期関数を収集し、それらを同時に実行します。この方法とは別に、次のに示すように、awaitキーワードを直接使用することもできます。

import asyncio

async def hello():
    await asyncio.sleep(1)
    return "hello leapcell"

async def example():
    result = await hello()
    print(result)

asyncio.run(example())

出力:

hello leapcell

上記のexample関数では、awaitを使用して非同期関数の結果を待ちます。結果が返された後、コンソールに出力されます。この方法は、コードがawaitステートメントで結果が返されるまで待機してから、次のコードの実行を継続するため、実際には順番に実行されます。

ここで待機しないとどうなりますか？ result = hello()を使用すると、hello()内のコードは実行されず、返されるresultはコルーチンオブジェクトになります。

asyncio.create_task()

上記の方法に加えて、より柔軟な方法があります。それはasyncio.create_task()を使用することです。このメソッドはタスクを作成し、バックグラウンドですぐに実行します。この時点で、メイン関数は他の操作を実行できます。非同期タスクの結果を取得する必要がある場合は、awaitを使用して取得します。次に示すように。

import asyncio

async def number():
    await asyncio.sleep(1)
    return 1

async def float_num():
    await asyncio.sleep(1)
    return 1.0

async def example():
    n = asyncio.create_task(number())
    f = asyncio.create_task(float_num())
    print("do something...")

    print(await n)
    print(await f)

asyncio.run(example())

出力:

do something...
1
1.0

上記の出力から、create_taskが最初にタスクを作成して開始することがわかります。この時点で、メイン関数はブロックされず、次のコードの実行に進みます。非同期関数の結果が必要な場合は、await nを呼び出して結果を取得します。このようにして、時間のかかるタスクを非同期コードに入れて実行し、必要に応じてこれらの非同期関数の結果を取得できます。

注：create_taskを使用して非同期関数を呼び出すことは、通常の関数のように呼び出すこととは異なります。通常関数呼び出しメソッドnumber()を使用すると、関数は実行されません。ただし、create_taskを使用して非同期関数を呼び出すと、関数はすぐに実行されます。awaitを使用して結果を取得しなくても、関数はメイン関数が終了する前に実行を完了します。

Semaphore

asyncio.Semaphoreは、Pythonのasyncioライブラリの同期プリミティブであり、共有リソースへのアクセスを制御します。これは非同期プログラミングで非常に役立ち、特定のリソースに同時にアクセスできるコルーチンの数を制限できます。次のコードに示すように：

import asyncio
import aiohttp

async def fetch(url, session, semaphore):
    async with semaphore:
        print(f"Fetching {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",
        "http://example.edu",
        "http://example.io",
    ]

    semaphore = asyncio.Semaphore(2)  # 同時リクエストの数を2に制限します
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(url, session, semaphore) for url in urls]
        responses = await asyncio.gather(*tasks)

    for url, response in zip(urls, responses):
        print(f"URL: {url}, Response length: {len(response)}")

asyncio.run(main())

上記のコードでは、asyncio.Semaphore(2)が作成され、同時リクエストの数を2に制限しています。非同期関数fetchでは、async with semaphoreを使用してセマフォを取得および解放します。入る前に、acquire()メソッドが自動的に呼び出されてセマフォが取得され、withブロックを終了するときに、release()メソッドが呼び出されてセマフォが解放されます。 Semaphoreを使用すると、同時リクエストの数を制御して、サーバーへの過度の圧力を防ぐことができます。これは、データベース接続などの限られたリソースを扱う場合に非常に役立ちます。同時に、システムパフォーマンスを最適化し、同時実行のバランスを見つけることができます。

Semaphoreの原則

内部的には、カウンターが維持されます。カウンターがゼロより大きい場合は、アクセスが許可されます。ゼロに等しい場合は、アクセスが禁止されます。カウンターを取得および解放する方法は、それぞれacquire()とrelease()です。初期化中に初期カウンター値を指定する必要があります。次に、コード内のカウンター値を制御することにより、同時リクエストの数を制御します。

マルチスレッディング

マルチスレッディングは、従来の同時タスク実行の方法であり、I/Oバウンドのタスクに適しています。次の例に示すように：

import threading
import time

def worker(name):
    print(f"Worker {name} starting")
    time.sleep(2)  # 時間のかかる操作をシミュレートします
    print(f"Worker {name} finished")

def main():
    threads = []
    for i in range(3):
        t = threading.Thread(target=worker, args=(i,))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    print("All workers finished")

if __name__ == "__main__":
    main()

マルチスレッディングのt.join()は、3つのスレッドの完了を待機するために使用されます。 join()メソッドがスレッドまたはプロセスオブジェクトで呼び出されると、呼び出しスレッド（通常はメインスレッド）は、join()が呼び出されたスレッドまたはプロセスが実行を終了するまでブロックされます。

マルチプロセッシング

マルチプロセッシングはCPUバウンドのタスクに適しており、マルチコアプロセッサを最大限に活用できます。以下に示すように：

import multiprocessing
import time

def worker(name):
    print(f"Worker {name} starting")
    time.sleep(2)  # 時間のかかる操作をシミュレートします
    print(f"Worker {name} finished")

if __name__ == "__main__":
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("All workers finished")

結論

上記の非同期メソッドに加えて、コールバック関数やGeventのようなサードパーティライブラリを使用するなど、Pythonには他の非同期アプローチがあります。各メソッドには、独自の利点と制限があります。たとえば、スレッドはI/Oバウンドのタスクに適していますが、GIL（グローバルインタープリターロック）によって制限されています。マルチプロセッシングはCPUバウンドのタスクに適していますが、メモリオーバーヘッドが高くなります。サードパーティライブラリは、特殊な機能と最適化を提供しますが、プロジェクトの複雑さを増す可能性があります。対照的に、async/await構文は、より現代的で読みやすい非同期プログラミングの方法を提供し、現在、Pythonで非同期操作を処理するために推奨されている方法です。