目錄

一、上下文管理器

1.1 定義

1.2 特點

1.3 適用場景

1.4 具體實現

1.5 具體實例

1.5.1 文件管理器

1.5.2 線程鎖釋放資源

二、異步I/O

2.1 定義

2.2 特點

2.3 實現方式

2.4 適用場景?

高并發網絡服務：Web服務器、API服務等需要處理大量并發連接

2.5 具體實現

2.6?await關鍵字

三、總結

一、上下文管理器

1.1 定義

上下文管理器（Context Manager）是Python中一種特殊對象，它實現了enter和exit方法，用于在with語句中進行資源管理。它確保在進入和退出代碼塊時執行特定的初始化和清理操作。
具體實現enter和exit方法的方法：

class FileManager:
def __init__(self, filename, mode):
self.filename = filename
self.mode = mode
def __enter__(self):
self.file = open(self.filename, self.mode)
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
self.file.close()
# 使用
with FileManager('example.txt', 'r') as f:
content = f.read()

從具體的實現方式來看，上下文管理器是一種支持 with 語句的對象，用于管理資源（如文件、網絡連接等）的獲取和釋放。

1.2 特點

• 自動資源管理：通過with語句使用，確保資源正確分配和釋放
• 異常安全：即使在執行過程中發生異常，也能確保清理代碼被執行
• 簡潔語法：使用with語句簡化代碼，提高可讀性
• 明確的生命周期：明確定義了進入和退出代碼塊時的行為

1.3 適用場景

1. 文件操作（自動打開和關閉文件）
2. 數據庫連接管理
3. 鎖定和解鎖資源
4. 臨時狀態更改（如臨時改變工作目錄）

1.4 具體實現

上面通過enter和exit方法實現只是一種，也能夠通過yield在函數中實現

from contextlib import contextmanager@contextmanager
def my_context_manager(message):print(f"Entering context: {message}")try:# yield 之前的代碼相當于 __enter__yield messagefinally:# yield 之后的代碼相當于 __exit__print(f"Exiting context: {message}")# 使用示例
with my_context_manager("test") as msg:print(f"Inside context with message: {msg}")

1.5 具體實例

1.5.1 文件管理器

class FileManager:def __init__(self, filename, mode):self.filename = filenameself.mode = modeself.file = Nonedef __enter__(self):print(f"Opening file: {self.filename}")self.file = open(self.filename, self.mode)return self.filedef __exit__(self, exc_type, exc_value, traceback):print(f"Closing file: {self.filename}")if self.file:self.file.close()# 處理異常if exc_type is not None:print(f"Exception occurred: {exc_value}")return False  # 不抑制異常# 使用示例
with FileManager("test.txt", "w") as f:f.write("Hello, World!")

1.5.2 線程鎖釋放資源

import threading
# 創建?個鎖對象，?于在多線程中同步對共享資源的訪問
lock = threading.Lock()# 定義線程函數，該函數將計數器增加n次
def thread_function(n):global counterfor _ in range(n):# 打印當前線程將計數器增加到的值print(f'線程 {threading.current_thread().name} 將計數器增加到 {counter}')# 獲取鎖，確保同?時間只有?個線程在修改counterlock.acquire()try:# 嘗試增加計數器counter += 1finally:# 釋放鎖，允許其他線程進?操作lock.release()# 初始化全局計數器
counter = 0
# 初始化線程列表
threads = []# 創建并啟動5個線程，每個線程都會執?thread_function
for i in range(5):# 創建線程t = threading.Thread(target=thread_function, args=(i,))# 將線程添加到線程列表中threads.append(t)# 啟動線程t.start()# 等待所有線程完成執?for t in threads:t.join()
# 打印最終的計數器值
print(f'最終的計數器值: {counter}')

二、異步I/O

2.1 定義

異步 I/O（Asynchronous I/O）指發起 I/O 操作后不必等待其完成，內核在后臺把數據準備好或寫入完成后，再通過回調、事件通知或 Future 等機制通知用戶程序；整個過程不會阻塞調用線程，線程可立即轉去執行其他任務，從而用極少的線程支撐海量并發。

2.2 特點

• 非阻塞：調用立即返回，線程不掛起

• 事件驅動：完成通知通過回調、Future、epoll、kqueue、IOCP 等事件機制觸發。

• 高吞吐、低延遲：單線程即可管理成千上萬個并發連接，節省上下文切換和內存開銷。

• 編程模型復雜：需要處理回調地獄、狀態機、錯誤傳播、背壓等問題；現代語言用 Promise/Future/協程/actor 封裝。

2.3 實現方式

1.async/await語法：Python 3.5+引入的關鍵字，用于定義協程函數
2.asyncio庫：Python標準庫，提供事件循環、協程、任務等核心功能
3.第三方庫支持：如aiohttp（HTTP客戶端/服務端）、aiomysql（MySQL數據庫）等
?

2.4 適用場景?

• 高并發網絡服務：Web服務器、API服務等需要處理大量并發連接

• I/O密集型應用：文件操作、數據庫查詢、網絡請求等耗時操作
• 實時數據處理：實時消息處理、流數據處理等
• 爬蟲應用：需要同時發起多個網絡請求的爬蟲程序

2.5 具體實現

基礎異步函數：

import asyncio
import timeasync def say_after(delay, what):await asyncio.sleep(delay)print(what)async def main():print(f"started at {time.strftime('%X')}")await say_after(1, 'hello')await say_after(2, 'world')print(f"finished at {time.strftime('%X')}")# 運行異步函數
asyncio.run(main())

2.6?await關鍵字

Python中async/await的工作原理：

import asyncioasync def say_after(delay, what):await asyncio.sleep(delay)  # 暫停執行delay秒print(what)async def main():print('開始')await say_after(1, 'hello')  # 等待1秒后打印helloawait say_after(2, 'world')  # 再等待2秒后打印worldprint('結束')# 運行主協程
asyncio.run(main())