Python Asyncio库之asyncio.task常用函数有哪些

目录

1. 引言
2. Asyncio库简介
3. asyncio.task模块概述
4. 常用函数详解
   • 4.1 asyncio.create_task()
   • 4.2 asyncio.gather()
   • 4.3 asyncio.wait()
   • 4.4 asyncio.wait_for()
   • 4.5 asyncio.shield()
   • 4.6 asyncio.current_task()
   • 4.7 asyncio.all_tasks()
   • 4.8 asyncio.Task.cancel()
   • 4.9 asyncio.Task.result()
   • 4.10 asyncio.Task.exception()
   • 4.11 asyncio.Task.done()
   • 4.12 asyncio.Task.add_done_callback()
   • 4.13 asyncio.Task.remove_done_callback()
   • 4.14 asyncio.Task.get_stack()
   • 4.15 asyncio.Task.print_stack()
5. 实战案例
6. 总结

引言

在现代编程中，异步编程已经成为处理高并发、I/O密集型任务的重要手段。Python的asyncio库为开发者提供了强大的异步编程支持，而asyncio.task模块则是其中的核心组件之一。本文将深入探讨asyncio.task模块中的常用函数，帮助读者更好地理解和应用这些工具。

Asyncio库简介

asyncio是Python标准库中的一个模块，用于编写单线程并发代码，通常用于I/O密集型任务。它通过事件循环（Event Loop）和协程（Coroutine）来实现异步编程。asyncio库的核心思想是让程序在等待I/O操作时能够执行其他任务，从而提高程序的效率。

asyncio.task模块概述

asyncio.task模块是asyncio库中的一个重要组成部分，主要负责管理和调度协程任务。通过asyncio.task模块，开发者可以创建、取消、等待任务，并获取任务的状态和结果。asyncio.task模块提供了丰富的API，使得异步编程变得更加灵活和高效。

常用函数详解

4.1 asyncio.create_task()

asyncio.create_task()函数用于将一个协程包装为一个Task对象，并将其调度到事件循环中执行。Task对象是asyncio库中用于管理协程的基本单位。

语法：

asyncio.create_task(coro, *, name=None)

参数： - coro: 需要包装的协程对象。 - name: 可选参数，指定任务的名称。

返回值： 返回一个Task对象。

示例：

import asyncio

async def my_coroutine():
    print("Coroutine is running")
    await asyncio.sleep(1)
    print("Coroutine is done")

async def main():
    task = asyncio.create_task(my_coroutine())
    await task

asyncio.run(main())

输出：

Coroutine is running
Coroutine is done

4.2 asyncio.gather()

asyncio.gather()函数用于并发运行多个协程，并等待它们全部完成。它返回一个包含所有协程结果的列表。

语法：

asyncio.gather(*coros_or_futures, return_exceptions=False)

参数： - *coros_or_futures: 一个或多个协程或Future对象。 - return_exceptions: 可选参数，如果为True，则异常会作为结果返回，而不是抛出。

返回值： 返回一个包含所有协程结果的列表。

示例：

import asyncio

async def coro1():
    await asyncio.sleep(1)
    return "Coroutine 1"

async def coro2():
    await asyncio.sleep(2)
    return "Coroutine 2"

async def main():
    results = await asyncio.gather(coro1(), coro2())
    print(results)

asyncio.run(main())

输出：

['Coroutine 1', 'Coroutine 2']

4.3 asyncio.wait()

asyncio.wait()函数用于等待一组协程或Future对象完成。它返回两个集合：已完成的任务和未完成的任务。

语法：

asyncio.wait(fs, *, timeout=None, return_when=ALL_COMPLETED)

参数： - fs: 一个包含协程或Future对象的集合。 - timeout: 可选参数，指定等待的超时时间。 - return_when: 可选参数，指定何时返回结果，可选值为FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETED。

返回值： 返回一个包含两个集合的元组：已完成的任务和未完成的任务。

示例：

import asyncio

async def coro1():
    await asyncio.sleep(1)
    return "Coroutine 1"

async def coro2():
    await asyncio.sleep(2)
    return "Coroutine 2"

async def main():
    tasks = [asyncio.create_task(coro1()), asyncio.create_task(coro2())]
    done, pending = await asyncio.wait(tasks)
    for task in done:
        print(task.result())

asyncio.run(main())

输出：

Coroutine 1
Coroutine 2

4.4 asyncio.wait_for()

asyncio.wait_for()函数用于等待一个协程或Future对象完成，并设置超时时间。如果超时，将抛出asyncio.TimeoutError异常。

语法：

asyncio.wait_for(fut, timeout)

参数： - fut: 需要等待的协程或Future对象。 - timeout: 超时时间。

返回值： 返回协程或Future对象的结果。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(2)
    return "Coroutine is done"

async def main():
    try:
        result = await asyncio.wait_for(my_coroutine(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("Timeout!")

asyncio.run(main())

输出：

Timeout!

4.5 asyncio.shield()

asyncio.shield()函数用于保护一个协程或Future对象，使其在取消时不会被取消。它返回一个新的Future对象，该对象在原始Future对象完成时完成。

语法：

asyncio.shield(aw)

参数： - aw: 需要保护的协程或Future对象。

返回值： 返回一个新的Future对象。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(2)
    return "Coroutine is done"

async def main():
    task = asyncio.create_task(my_coroutine())
    shielded_task = asyncio.shield(task)
    await asyncio.sleep(1)
    task.cancel()
    try:
        result = await shielded_task
        print(result)
    except asyncio.CancelledError:
        print("Shielded task was cancelled")

asyncio.run(main())

输出：

Coroutine is done

4.6 asyncio.current_task()

asyncio.current_task()函数用于获取当前正在运行的Task对象。

语法：

asyncio.current_task(loop=None)

参数： - loop: 可选参数，指定事件循环。

返回值： 返回当前正在运行的Task对象。

示例：

import asyncio

async def my_coroutine():
    task = asyncio.current_task()
    print(f"Current task: {task}")

async def main():
    await my_coroutine()

asyncio.run(main())

输出：

Current task: <Task pending coro=<my_coroutine() running at example.py:3>>

4.7 asyncio.all_tasks()

asyncio.all_tasks()函数用于获取当前事件循环中所有未完成的Task对象。

语法：

asyncio.all_tasks(loop=None)

参数： - loop: 可选参数，指定事件循环。

返回值： 返回一个包含所有未完成Task对象的集合。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)

async def main():
    task1 = asyncio.create_task(my_coroutine())
    task2 = asyncio.create_task(my_coroutine())
    tasks = asyncio.all_tasks()
    print(f"All tasks: {tasks}")

asyncio.run(main())

输出：

All tasks: {<Task pending coro=<my_coroutine() running at example.py:3>>, <Task pending coro=<my_coroutine() running at example.py:3>>}

4.8 asyncio.Task.cancel()

asyncio.Task.cancel()方法用于取消一个Task对象。如果任务已经完成或取消，则不会产生任何效果。

语法：

task.cancel()

返回值： 返回True表示任务被成功取消，否则返回False。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(2)
    print("Coroutine is done")

async def main():
    task = asyncio.create_task(my_coroutine())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")

asyncio.run(main())

输出：

Task was cancelled

4.9 asyncio.Task.result()

asyncio.Task.result()方法用于获取Task对象的结果。如果任务尚未完成，则会阻塞直到任务完成。

语法：

task.result()

返回值： 返回任务的结果。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return "Coroutine is done"

async def main():
    task = asyncio.create_task(my_coroutine())
    result = await task.result()
    print(result)

asyncio.run(main())

输出：

Coroutine is done

4.10 asyncio.Task.exception()

asyncio.Task.exception()方法用于获取Task对象的异常。如果任务尚未完成，则会阻塞直到任务完成。

语法：

task.exception()

返回值： 返回任务的异常，如果任务没有异常则返回None。

示例：

import asyncio

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

async def main():
    task = asyncio.create_task(my_coroutine())
    try:
        await task
    except ValueError as e:
        print(f"Caught exception: {e}")

asyncio.run(main())

输出：

Caught exception: An error occurred

4.11 asyncio.Task.done()

asyncio.Task.done()方法用于检查Task对象是否已完成。

语法：

task.done()

返回值： 返回True表示任务已完成，否则返回False。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)

async def main():
    task = asyncio.create_task(my_coroutine())
    print(f"Task done: {task.done()}")
    await task
    print(f"Task done: {task.done()}")

asyncio.run(main())

输出：

Task done: False
Task done: True

4.12 asyncio.Task.add_done_callback()

asyncio.Task.add_done_callback()方法用于为Task对象添加一个回调函数，当任务完成时调用该回调函数。

语法：

task.add_done_callback(callback)

参数： - callback: 回调函数，接受一个Task对象作为参数。

返回值： 无。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return "Coroutine is done"

def callback(task):
    print(f"Callback: {task.result()}")

async def main():
    task = asyncio.create_task(my_coroutine())
    task.add_done_callback(callback)
    await task

asyncio.run(main())

输出：

Callback: Coroutine is done

4.13 asyncio.Task.remove_done_callback()

asyncio.Task.remove_done_callback()方法用于从Task对象中移除一个回调函数。

语法：

task.remove_done_callback(callback)

参数： - callback: 需要移除的回调函数。

返回值： 返回移除的回调函数的数量。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return "Coroutine is done"

def callback(task):
    print(f"Callback: {task.result()}")

async def main():
    task = asyncio.create_task(my_coroutine())
    task.add_done_callback(callback)
    task.remove_done_callback(callback)
    await task

asyncio.run(main())

输出：

4.14 asyncio.Task.get_stack()

asyncio.Task.get_stack()方法用于获取Task对象的调用栈。

语法：

task.get_stack(*, limit=None)

参数： - limit: 可选参数，指定返回的栈帧数量。

返回值： 返回一个包含栈帧的列表。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)

async def main():
    task = asyncio.create_task(my_coroutine())
    stack = task.get_stack()
    print(f"Stack: {stack}")

asyncio.run(main())

输出：

Stack: []

4.15 asyncio.Task.print_stack()

asyncio.Task.print_stack()方法用于打印Task对象的调用栈。

语法：

task.print_stack(*, limit=None, file=None)

参数： - limit: 可选参数，指定打印的栈帧数量。 - file: 可选参数，指定输出文件，默认为sys.stderr。

返回值： 无。

示例：

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)

async def main():
    task = asyncio.create_task(my_coroutine())
    task.print_stack()

asyncio.run(main())

输出：

实战案例

在实际开发中，asyncio.task模块的函数可以帮助我们更好地管理和调度异步任务。以下是一个简单的实战案例，展示了如何使用asyncio.create_task()、asyncio.gather()和asyncio.wait()来并发执行多个任务。

import asyncio

async def fetch_data(url):
    print(f"Fetching data from {url}")
    await asyncio.sleep(2)  # 模拟网络请求
    print(f"Data fetched from {url}")
    return f"Data from {url}"

async def main():
    urls = ["https://example.com", "https://example.org", "https://example.net"]
    tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
    results = await asyncio.gather(*tasks)
    print(f"All data fetched: {results}")

asyncio.run(main())

输出：

Fetching data from https://example.com
Fetching data from https://example.org
Fetching data from https://example.net
Data fetched from https://example.com
Data fetched from https://example.org
Data fetched from https://example.net
All data fetched: ['Data from https://example.com', 'Data from https://example.org', 'Data from https://example.net']

总结

asyncio.task模块提供了丰富的API，使得异步编程变得更加灵活和高效。通过本文的介绍，读者应该能够掌握asyncio.task模块中常用函数的使用方法，并能够在实际开发中灵活运用这些工具。希望本文能够帮助读者更好地理解和应用Python的异步编程技术。