Python協(xié)程怎樣處理異常情況

在Python中，協(xié)程（coroutine）是一種特殊的函數(shù)，可以在執(zhí)行過(guò)程中暫停和恢復(fù)。處理協(xié)程中的異常情況需要使用asyncio庫(kù)中的一些功能。以下是一些處理協(xié)程異常的方法：

1. 使用try-except語(yǔ)句捕獲異常：

在協(xié)程內(nèi)部使用try-except語(yǔ)句捕獲異常，就像在普通函數(shù)中一樣。當(dāng)異常發(fā)生時(shí)，它會(huì)被捕獲并存儲(chǔ)在except子句中。

import asyncio

async def my_coroutine():
    try:
        # 你的協(xié)程代碼
    except Exception as e:
        print(f"捕獲到異常: {e}")

asyncio.run(my_coroutine())

2. 使用asyncio.gather處理多個(gè)協(xié)程的異常：

asyncio.gather函數(shù)可以同時(shí)運(yùn)行多個(gè)協(xié)程，并收集它們的結(jié)果。如果其中一個(gè)協(xié)程引發(fā)異常，gather會(huì)立即停止執(zhí)行剩余的協(xié)程，并將異常傳遞給return_exceptions參數(shù)。

import asyncio

async def my_coroutine(num):
    if num == 2:
        raise ValueError("這是一個(gè)故意引發(fā)的異常")
    return f"協(xié)程 {num} 完成"

async def main():
    coroutines = [my_coroutine(i) for i in range(1, 4)]
    results = await asyncio.gather(*coroutines, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            print(f"捕獲到異常: {result}")
        else:
            print(result)

asyncio.run(main())

3. 使用asyncio.create_task和await處理異常：

當(dāng)你使用asyncio.create_task創(chuàng)建一個(gè)任務(wù)時(shí)，可以使用await關(guān)鍵字等待協(xié)程完成。如果協(xié)程引發(fā)異常，它會(huì)被捕獲并存儲(chǔ)在asyncio.Task對(duì)象的exception()方法中。

import asyncio

async def my_coroutine(num):
    if num == 2:
        raise ValueError("這是一個(gè)故意引發(fā)的異常")
    return f"協(xié)程 {num} 完成"

async def main():
    task = asyncio.create_task(my_coroutine(2))
    try:
        result = await task
    except Exception as e:
        print(f"捕獲到異常: {e}")
    else:
        print(result)

asyncio.run(main())

這些方法可以幫助你處理Python協(xié)程中的異常情況。在實(shí)際編程中，你可能需要根據(jù)具體需求選擇合適的方法。