Python?Asyncio庫之asyncio.task常用函數有哪些

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的異步編程技術。