Asyncio を使用したタスクの作成と管理

Asyncio を使用すると、開発者は手間をかけずに Python で非同期プログラムを作成できます。このモジュールは、非同期タスクのさまざまな方法も提供しますが、実行方法が多数あるため、どれを使用すればよいか混乱する可能性があります。

この記事では、asyncio を使用してタスクを作成および管理できるさまざまな方法について説明します。

非同期タスクとは何ですか?

asyncio では、タスク はコルーチンをラップし、イベント ループ内で実行するようにスケジュールするオブジェクトです。簡単に言えば、タスクはコルーチンを他のタスクと同時に実行する方法です。タスクが作成されると、イベント ループによってタスクが実行され、必要に応じて一時停止および再開され、他のタスクが実行できるようになります。

Asyncio タスクを作成および管理する方法

ここで、タスクを作成および管理する方法について説明します。まず、asyncio を使用して Python でタスクを作成するには、次の引数を取る asyncio.create_task メソッドを使用します。

• coro (必須): スケジュールされるコルーチン オブジェクト。これは、非同期で実行する関数です。

• name (オプション): デバッグやロギングの目的で役立つタスクの名前。このパラメータには文字列を割り当てることができます。

  • Task.set_name(name) と Task.get_name() を使用して、後で名前を設定または取得することもできます。

• context (オプション): Python 3.11 で導入され、タスクのコンテキスト変数を設定するために使用され、タスクのローカル ストレージが有効になります。これはスレッドローカル ストレージに似ていますが、非同期タスク用です。

  • この引数は、コンテキスト管理を必要とする高度なシナリオを扱っている場合を除き、一般的には使用されません。

これは asyncio.create_task の使用例です:

import asyncio

# Define a coroutine
async def greet(name):
    await asyncio.sleep(1)  # Simulate an I/O-bound operation
    print(f"Hello, {name}!")

async def main():
    # Create tasks
    task1 = asyncio.create_task(greet("Alice"), name="GreetingAlice")
    task2 = asyncio.create_task(greet("Bob"), name="GreetingBob")

    # Check task names
    print(f"Task 1 name: {task1.get_name()}")
    print(f"Task 2 name: {task2.get_name()}")

    # Wait for both tasks to complete
    await task1
    await task2

# Run the main function
asyncio.run(main())

タスクを作成すると、次のような多くのメソッドを実行できます。

• .cancel(): タスクをキャンセルします。

• .add_done_callback(cb): タスクの完了時に実行されるコールバック関数を追加します。

• .done(): タスクが完了したかどうかを確認します。

• .result(): タスクの完了後にその結果を取得します。

タスクの作成方法を理解したところで、1 つまたは複数のタスクの待機を処理する方法を見てみましょう。

タスクの完了を待っています

このセクションでは、1 つまたは複数のタスクのタスク完了を待つ方法について説明します。非同期プログラミングは、非同期タスクが実行されていればプログラムの実行を継続できるという事実に基づいています。フローをより適切に制御し、プログラムの実行を安全に続行する前に作業できる結果が得られることを確認したい場合があるかもしれません。

単一のタスクの完了を待つには、asyncio.wait_for を使用できます。 2 つの引数を取ります:

• awaitable (必須): これは、待機するコルーチン、タスク、または将来です。コルーチン関数呼び出し、asyncio.Task、asyncio.Future など、待機できる任意のオブジェクトを指定できます。

• timeout (オプション): これは、aw が完了するまで待機する最大秒数を指定します。タイムアウトに達しても awaitable が完了していない場合、asyncio.wait_for は TimeoutError を発生させます。タイムアウトが [なし] に設定されている場合、関数は awaitable が完了するまで無期限に待機します。

このメソッドが使用される例を次に示します:

import asyncio

async def slow_task():
    print("Task started...")
    await asyncio.sleep(5)  # Simulating a long-running task
    print("Task finished!")
    return "Completed"

async def main():
    try:
        # Wait for slow_task to finish within 2 seconds
        result = await asyncio.wait_for(slow_task(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("The task took too long and was canceled!")

asyncio.run(main())

上記のコードでは、slow_task() は 5 秒間スリープすることで長時間実行タスクをシミュレートするコルーチンです。 asyncio.wait_for(slow_task(), timeout=2) 行はタスクが完了するのを待ちますが、待ち時間を 2 秒に制限しており、タスクに時間がかかるためタイムアウトが発生します。タイムアウトを超えると、TimeoutError が発生し、タスクがキャンセルされ、タスクに時間がかかりすぎたことを示すメッセージを出力することで例外が処理されます。

複数またはグループのタスクが完了するのを待つこともできます。これは、asyncio.wait、asyncio.gather、または asyncio.as_completed を使用して可能です。それぞれの方法を見てみましょう。

asyncio.wait

asyncio.wait メソッドは、タスクのコレクションを待機し、完了したタスク用と保留中のタスク用の 2 つのセットを返します。次の引数を受け取ります:

• aws (必須、awaitable の反復可能): 待機するコルーチン オブジェクト、タスク、または Future のコレクション。

• timeout (float または None、オプション): 待機する最大秒数。指定しない場合は、無期限に待機します。

• return_when (定数、オプション): asyncio.wait が返されるタイミングを指定します。オプションには以下が含まれます:

  • asyncio.ALL_COMPLETED (default): Returns when all tasks are complete.
  • asyncio.FIRST_COMPLETED: Returns when the first task is completed.
  • asyncio.FIRST_EXCEPTION: Returns when the first task raises an exception.

Let's see how it is used in an example.

import asyncio
import random

async def task():
    await asyncio.sleep(random.uniform(1, 3))

async def main():
    tasks = [asyncio.create_task(task()) for _ in range(3)]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print(f"Done tasks: {len(done)}, Pending tasks: {len(pending)}")

asyncio.run(main())

In the code above, asyncio.wait waits for a group of tasks and returns two sets: one with completed tasks and another with those still pending. You can control when it returns, such as after the first task is completed or after all tasks are done. In the example, asyncio.wait returns when the first task is completed, leaving the rest in the pending set.

asyncio.gather

The asyncio.gather method runs multiple awaitable objects concurrently and returns a list of their results, optionally handling exceptions. Let's see the arguments it takes.

• *aws (required, multiple awaitables): A variable number of awaitable objects (like coroutines, tasks, or futures) to run concurrently.

• return_exceptions (bool, optional): If True, exceptions in the tasks will be returned as part of the results list instead of being raised.

Let's see how it can be used in an example.

import asyncio
import random

async def task(id):
    await asyncio.sleep(random.uniform(1, 3))
    return f"Task {id} done"

async def main():
    results = await asyncio.gather(task(1), task(2), task(3))
    print(results)

asyncio.run(main())

In the code above, asyncio.gather runs multiple awaitable objects concurrently and returns a list of their results in the order they were passed in. It allows you to handle exceptions gracefully if return_exceptions is set to True. In the example, three tasks are run simultaneously, and their results are returned in a list once all tasks are complete.

asyncio.as_completed

The asyncio.as_completed method is used to return an iterator that yields tasks as they are completed, allowing results to be processed immediately. It takes the following arguments:

• aws (iterable of awaitables): A collection of coroutine objects, tasks, or futures.

• timeout (float or None, optional): The maximum number of seconds to wait for tasks to complete. If not provided, it waits indefinitely.

Example

import asyncio
import random

async def task(id):
    await asyncio.sleep(random.uniform(1, 3))
    return f"Task {id} done"

async def main():
    tasks = [task(i) for i in range(3)]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(result)

asyncio.run(main())

In the example above, asyncio.as_completed returns an iterator that yields results as each task completes, allowing you to process them immediately. This is useful when you want to handle results as soon as they're available, rather than waiting for all tasks to finish. In the example, the tasks are run simultaneously, and their results are printed as each one finishes, in the order they complete.

So to make a summary, you use:

• asyncio.wait: when you need to handle multiple tasks and want to track which tasks are completed and which are still pending. It's useful when you care about the status of each task separately.

• asyncio.gather: when you want to run multiple tasks concurrently and need the results in a list, especially when the order of results matters or you need to handle exceptions gracefully.

• asyncio.as_completed: when you want to process results as soon as each task finishes, rather than waiting for all tasks to complete. It’s useful for handling results in the order they become available.

However, these methods don't take atomic task management with built-in error handling. In the next section, we will see about asyncio.TaskGroup and how to use it to manage a group of tasks.

asyncio.TaskGroup

asyncio.TaskGroup is a context manager introduced in Python 3.11 that simplifies managing multiple tasks as a group. It ensures that if any task within the group fails, all other tasks are canceled, providing a way to handle complex task management with robust error handling. The class has one method called created_task used to create and add tasks to the task group. You pass a coroutine to this method, and it returns an asyncio.Task object that is managed by the group.

Here is an example of how it is used:

import asyncio

async def task1():
    await asyncio.sleep(1)
    return "Task 1 done"

async def task2():
    await asyncio.sleep(2)
    return "Task 2 done"

async def task_with_error():
    await asyncio.sleep(1)
    raise ValueError("An error occurred")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(task1())
            task2 = tg.create_task(task2())
            error_task = tg.create_task(task_with_error())
    except Exception as e:
        print(f"Error: {e}")

    # Print results from completed tasks
    print("Task 1 result:", task1.result())
    print("Task 2 result:", task2.result())

asyncio.run(main())

asyncio.TaskGroup manages multiple tasks and ensures that if any task fails, all other tasks in the group are canceled. In the example, a task with an error causes the entire group to be canceled, and only the results of completed tasks are printed.

Usage for this can be in web scraping. You can use asyncio.TaskGroup to handle multiple concurrent API requests and ensure that if any request fails, all other requests are canceled to avoid incomplete data.

We are at the end of the article and we have learned the multiple methods asyncio provides to create and manage tasks. Here is a summary of the methods:

• asyncio.wait_for: Wait for a task with a timeout.

• asyncio.wait: Wait for multiple tasks with flexible completion conditions.

• asyncio.gather: Aggregate multiple tasks into a single awaitable.

• asyncio.as_completed: 完了したタスクを処理します。

• asyncio.TaskGroup: 失敗時に自動的にキャンセルされるタスクのグループを管理します。

結論

非同期プログラミングは、Python での同時タスクの処理方法を変革し、コードの効率と応答性を向上させます。この記事では、単純なタイムアウトから高度なタスク グループまで、タスクを作成および管理するために asyncio が提供するさまざまな方法を説明しました。各メソッド (asyncio.wait_for、asyncio.wait、asyncio.gather、asyncio.as_completed、および asyncio.TaskGroup) をいつどのように使用するかを理解することは、非同期プログラミングの可能性を最大限に活用し、アプリケーションをより堅牢でスケーラブルにするのに役立ちます。

非同期プログラミングとより実践的な例についてさらに詳しく知りたい場合は、こちらの詳細ガイドを参照してください。

この記事を気に入っていただけた場合は、今後の最新情報を見逃さないように、ニュースレターの購読をご検討ください。

コーディングを楽しんでください!

以上がAsyncio を使用したタスクの作成と管理の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。