Python 异步之如何获取当前和正在运行任务详解

Python 异步之获取当前和正在运行任务详解

在 Python 异步编程中，我们经常需要获取当前的任务以及正在运行的任务。本文将介绍如何使用 asyncio 模块和相关方法来获取这些信息。

获取当前任务

要获取当前的任务，我们可以使用 asyncio.current_task() 方法。该方法返回当前协程对象的 Task 实例。在异步中，Task 是 asyncio 的核心概念，表示一个异步任务。

以下是一个示例代码：

import asyncio

async def my_coroutine():
    task = asyncio.current_task()
    print(task)

async def main():
    task1 = asyncio.create_task(my_coroutine())
    task2 = asyncio.create_task(my_coroutine())
    tasks = [task1, task2]
    await asyncio.gather(*tasks)

asyncio.run(main())

在上面的代码中，我们定义了 my_coroutine 函数，该函数使用 asyncio.current_task() 方法获取当前的任务，并打印出该任务。我们还定义了 main 函数，该函数使用 asyncio.create_task() 方法创建两个任务，并使用 asyncio.gather() 方法等待所有任务完成。

当我们运行该代码时，输出将是类似如下的内容：

<Task pending name='Task-1' coro=<my_coroutine() running at example.py:4>>
<Task pending name='Task-2' coro=<my_coroutine() running at example.py:4>>

我们可以看到，返回的任务对象包含任务的状态和名称。

获取正在运行的任务

要获取正在运行的任务，我们可以使用 asyncio.all_tasks() 方法。该方法返回当前事件循环中所有任务对象（包括已经完成和取消的任务），然后我们可以使用 Task 对象的状态属性来检查任务是否正在运行。

以下是一个示例代码：

import asyncio

async def my_coroutine():
    tasks = asyncio.all_tasks()
    for task in tasks:
        if task != asyncio.current_task() and not task.done():
            print("Running task:", task)
    await asyncio.sleep(1)

async def main():
    task1 = asyncio.create_task(my_coroutine())
    task2 = asyncio.create_task(my_coroutine())
    tasks = [task1, task2]
    await asyncio.gather(*tasks)

asyncio.run(main())

在上面的代码中，我们定义了 my_coroutine 函数，该函数使用 asyncio.all_tasks() 方法获取当前的所有任务，并使用循环遍历所有任务。我们使用 if 语句来检查任务是否正在运行，并使用 Task.done() 方法来检查任务是否已经完成。如果任务正在运行，则打印出任务对象。

当我们运行该代码时，输出将是类似如下的内容：

Running task: <Task pending name='Task-1' coro=<my_coroutine() running at example.py:4>>
Running task: <Task pending name='Task-2' coro=<my_coroutine() running at example.py:4>>
Running task: <Task pending name='Task-1' coro=<my_coroutine() running at example.py:4>>
Running task: <Task pending name='Task-2' coro=<my_coroutine() running at example.py:4>>
Running task: <Task pending name='Task-1' coro=<my_coroutine() running at example.py:4>>
Running task: <Task pending name='Task-2' coro=<my_coroutine() running at example.py:4>>
...

我们可以看到，我们成功地获取了所有正在运行的任务对象。由于我们在每个协程中使用了 asyncio.sleep 方法，因此任务对象将被多次打印。