Python非同步程式設計進階錯誤處理與效能最佳化

print("HTTP fetch error:", e)
return contents

async def coordinate_with_gather():
    tasks = [asyncio.create_task(task_worker(i)) for i in range(10)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for idx, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {idx} failed with error: {result}")
        else:
            print(f"Task {idx} succeeded with result: {result}")

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

async def perform_async_operations():
    # 標準的 asyncio 操作現在將使用 uvloop 事件迴圈
    await asyncio.sleep(0.1)
    print("Uvloop in action!")

if __name__ == '__main__':
    asyncio.run(perform_async_operations())

asyncio.run(debug=True)

# 使用 mocking 函式庫模擬非同步 I/O 行為和誘發的延遲

import asyncio

async def simple_worker(idx: int) -> str:
    await asyncio.sleep(1)
    result = f"Task {idx} finished"
    print(f"Task {idx} result: {result}")
    return result

async def coordinate_with_gather():
    tasks = [simple_worker(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    return results

if __name__ == '__main__':
    asyncio.run(coordinate_with_gather())

import asyncio
import random

async def variable_worker(task_id: int) -> str:
    await asyncio.sleep(random.uniform(0.1, 0.5))
    return f"Task {task_id} completed."

async def process_tasks_as_completed():
    tasks = [asyncio.create_task(variable_worker(i)) for i in range(10)]
    for finished in asyncio.as_completed(tasks):
        result = await finished
        print(result)

if __name__ == '__main__':
    asyncio.run(process_tasks_as_completed())

import asyncio

async def producer(queue: asyncio.Queue, count: int) -> None:
    for i in range(count):
        await asyncio.sleep(0.1)
        await queue.put(i)
        print(f"Produced {i}")
    await queue.put(None)  # Sentinel 表示完成

async def consumer(queue: asyncio.Queue) -> None:
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        await asyncio.sleep(0.15)
        print(f"Consumed {item}")
        queue.task_done()

async def coordinated_pipeline():
    queue = asyncio.Queue()
    prod = asyncio.create_task(producer(queue, 10))
    cons = asyncio.create_task(consumer(queue))
    await asyncio.gather(prod, cons)
    await queue.join()  # 確保所有任務被處理

if __name__ == '__main__':
    asyncio.run(coordinated_pipeline())

import asyncio

async def subtask(identifier: int) -> None:
    await asyncio.sleep(0.2)
    print(f"Subtask {identifier} completed.")

async def parent_task():
    async with asyncio.TaskGroup() as tg:
        for i in range(5):
            tg.create_task(subtask(i))

async def structured_concurrency_demo():
    try:
        await parent_task()
    except* Exception as e:
        print(f"Encountered exception group: {e}")

if __name__ == '__main__':
    asyncio.run(structured_concurrency_demo())

import asyncio

async def cancellable_worker(event: asyncio.Event, worker_id: int) -> None:
    try:
        for i in range(10):
            if event.is_set():
                print(f"Worker {worker_id} received cancellation signal.")
                return
            await asyncio.sleep(0.1)
            print(f"Worker {worker_id} task {i} completed.")
    except asyncio.CancelledError:
        print(f"Worker {worker_id} was cancelled.")
        raise

async def coordinator_with_cancellation():
    cancel_event = asyncio.Event()
    tasks = [asyncio.create_task(cancellable_worker(cancel_event, i)) for i in range(3)]
    await asyncio.sleep(0.5)
    cancel_event.set()
    await asyncio.gather(*tasks, return_exceptions=True)
    print("All workers have been signalled to terminate.")

if __name__ == '__main__':
    asyncio.run(coordinator_with_cancellation())

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

async def perform_work():
    # 執行非同步I/O密集型任務，效能得到增強
    await asyncio.sleep(0.1)
    return "工作完成"

if __name__ == '__main__':
    result = asyncio.run(perform_work())
    print(result)

#### 內容解密：
1. **`asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())`**：這行程式碼將事件迴圈策略設定為`uvloop`，從而利用其高效能特性。
2. **`perform_work`函式**：這是一個非同步函式，模擬了一項耗時0.1秒的非同步I/O操作。
3. **`asyncio.run(perform_work())`**：執行`perform_work`協程並取得其結果。

### 自定義排程策略

另一個先進的做法是建立和使用自定義排程策略。`asyncio`內建的排程通常依賴於FIFO或呼叫順序原則，但某些工作負載可以從優先或動態任務排程中受益。透過實作自定義排程器——例如，使用根據堆積的優先順序佇列——開發者可以根據執行時指標（如任務緊急程度、預估I/O延遲或依賴狀態）控制執行順序。這種方法在處理異構工作負載（任務完成時間差異很大）的系統中尤其有用。自定義排程器可以根據最近的效能動態調整優先順序，從而使系統能夠在回應速度和吞吐量之間取得平衡。

```python
import asyncio
import heapq

class PriorityTask:
    def __init__(self, priority: int, coro: asyncio.coroutine):
        self.priority = priority
        self.coro = coro
    
    def __lt__(self, other):
        return self.priority < other.priority

async def task_function(name: str, delay: float):
    await asyncio.sleep(delay)
    print(f"任務 {name} 在 {delay} 秒後完成。")
    return name

async def custom_scheduler():
    tasks_heap = []
    # 較低的數字值表示較高的優先順序
    heapq.heappush(tasks_heap, PriorityTask(1, task_function("A", 0.3)))
    heapq.heappush(tasks_heap, PriorityTask(3, task_function("B", 0.1)))
    heapq.heappush(tasks_heap, PriorityTask(2, task_function("C", 0.2)))
    
    results = []
    while tasks_heap:
        pt = heapq.heappop(tasks_heap)
        result = await pt.coro
        results.append(result)
    return results

if __name__ == '__main__':
    final_results = asyncio.run(custom_scheduler())
    print("優先順序排程器結果：", final_results)

#### 內容解密：
1. **`PriorityTask`類別**：封裝了一個具有優先順序的任務，用於在優先順序佇列中進行排序。
2. **`task_function`函式**：模擬一個延遲完成的非同步任務。
3. **`custom_scheduler`函式**：實作了一個自定義的優先順序排程器，使用堆積資料結構來管理任務的執行順序。
4. **`heapq.heappush`和`heapq.heappop`**：用於向堆積中新增任務和從堆積中彈出任務，以實作優先順序排序。

### 效能最佳化

效能最佳化還延伸到策略性地使用`asyncio.run_in_executor`來解除安裝阻塞或CPU密集型任務。在非同步I/O與計算密集型工作負載交織的系統中，將任務適當地分配到事件迴圈和外部執行器之間至關重要。先進的開發者可以透過這種方式確保系統資源得到有效利用，同時保持良好的回應性和吞吐量。