pythonで並行処理を理解する

並行処理を使わない場合

3つのURLに対してネットワークI/Oを発行する例を考えます。ここではsleep(2)を使ってネットワークI/Oを再現しています。

何も考えずに愚直に実装するとこのようになります。特徴的なのは、sleep(ネットワークI/O)の処理によってforループがブロックされるということです。そのため、処理には2*3で6sかかってしまいます。

import time


def fetch(url):
    # sleep中は、このプログラムのほかの処理も進まない。
    time.sleep(2)

    return url + " data"


def main():
    urls = [
        "https://example.com/1",
        "https://example.com/2",
        "https://example.com/3",
    ]

    for url in urls:
        # fetchが終わるまで次のループへ進まない。
        fetch(url)

main()

処理の流れを図で表現すると以下のようになります。各taskではネットワークI/O待ちが発生しているとして、それが完了しないとforループを回して次のtaskの実行に移れないということです。これでは非効率なので、後述する並行処理について検討する必要があります。

並行処理を使う場合

fetch()にて、await asyncio.sleep(2) を使うようにしています。asyncioによりsleep中(ネットワークI/O待ち中)において処理をブロックせず別の非同期タスクを実行できるようにします。

また、sleep(ネットワークI/O)の完了まで fetch() を一時停止し、その間に他タスクを実行可能にするawaitにより、sleep完了後にreturnするようにしています。

なお、asyncio.create_task() は、非同期関数の処理を「実行予定のタスク」としてイベントループへ登録する処理です。これにより、fetch(url) は他のタスクと並行して実行できる状態になります。イベントループは、asyncio.run(main())の実行により内部で作成されます。

import asyncio


async def fetch(url):
    # ネットワークからの応答待ちを再現。await中は、ほかのタスクを実行できる。
    await asyncio.sleep(2)

    return url + " のデータ"


async def main():
    urls = [
        "https://example.com/1",
        "https://example.com/2",
        "https://example.com/3",
    ]

    tasks = []
    for url in urls:
        tasks.append(asyncio.create_task(fetch(url)))

    await asyncio.gather(*tasks)


asyncio.run(main())

処理の流れを図で表現すると以下になります。先ほどとの違いとしては、各taskの実行開始タイミングは前のtaskの実行完了を待たないということです。並行処理により、ネットワークI/O待ちのタイミングでイベントループが他のtaskを実行できるようになっている点で効率的になっています。

forループを回して順番にtaskを作成し、1つのイベントループ上で切り替えながら実行するという点で並列処理ではないですが、実行開始のタイミングは先ほどのネットワークI/O待ちと比較するとメモリ上のループ処理で無視できる差になっています。