如何使用上下文管理器管理 Python 中的資源

在構建應用程序時，正確管理資源至關重要，以防止內存洩漏、確保正確的清理並保持應用程序的穩定性。上下文管理器為這種情況提供了一個完善的解決方案。上下文管理器通過自動化資源獲取和釋放過程來簡化資源管理。

什麼是上下文管理器?

上下文管理器本質上是一個對象，它描述了保護資源並在不再需要時釋放它們的策略。這些實體以連貫、直接且明確的格式呈現資源管理，從而提供了簡化資源管理的優勢。通過使用上下文管理器，可以減少冗餘代碼並增強代碼庫的易讀性。

軟件應用程序通常需要將數據記錄到文件中。在未使用上下文管理器的情況下，用戶必須手動管理日誌文件的打開和關閉。然而，通過使用上下文管理器，可以簡化建立和拆除日誌記錄資源的過程，確保日誌記錄操作的有效執行，同時還保持對相關任務的適當處理。

with 語句

在 Python 編程語言中使用 with 語句使開發人員能夠有效地利用上下文管理器的功能。儘管在執行相關代碼塊期間發生任何異常，此構造仍保證所有獲取的資源都被正確放棄並相應地實現其預期目的。

 with context_manager_expression as resource:
    # Code block that uses the resource
# Resource is automatically released when the block exits

通過使用“with”語句，上下文管理器被授予資源管理權限，從而將您的注意力集中到程序邏輯的複雜性上。

使用內置上下文管理器

Python 提供了內置的上下文管理器來滿足常見情況，例如使用 open() 函數處理文件以及通過 socket 模塊調節網絡連接。

使用 open() 處理文件

open() 函數充當內置上下文管理器，促進 Python 編程中與文件的交互。該實用程序通常用於讀取或寫入文件，並在執行時生成文件對象。其主要優點之一在於自動關閉文件，這有助於防止在託管代碼塊中工作時出現任何意外的數據損壞。

 with open('file.txt', 'r') as file:
    content = file.read()
    # Do something with content
# File is automatically closed after exiting the block

使用 socket() 進行網絡連接

socket 模塊為網絡套接字提供了一個上下文管理器，確保網絡連接的適當配置和處理，從而減輕與不正確終止連接相關的潛在安全風險。

 import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect(('localhost', 8080))
    # Send/receive data over the socket
# Socket is automatically closed after exiting the block

實現自定義上下文管理器

Python 允許通過基於類和基於函數的方法創建自定義上下文管理器。這些方法之間的選擇取決於當前情況的具體要求。在本節中，我們將研究如何使用這兩種方法來實現自定義上下文管理器。

使用基於類的方法的上下文管理器

在基於類的方法中，定義一個類，該類通過實現兩個特殊方法（稱為 \enter\ 和 \_\exit\ ）來體現管理資源的概念。這些神奇的或雙下劃線的方法有不同的目的；前者負責設置和產生所需的資源，而後者確保執行徹底的清理過程，無論執行期間可能出現任何潛在的異常。

 class CustomContext:
    def __enter__(self):
        # Acquire the resource
        return resource

    def __exit__(self, exc_type, exc_value, traceback):
        # Release the resource
        pass

為了在給定任務中同時有效地執行多個進程，有必要使用上下文管理器來簡化該進程，同時處理資源分配、協調和錯誤解決等問題。理想的解決方案應該自動生成、執行和無縫集成這些流程，確保資源的正確管理以及它們之間的同步。

 import multiprocessing
import queue

class ProcessPool:
    def __init__(self, num_processes):
        self.num_processes = num_processes
        self.processes = []

    def __enter__(self):
        self.queue = multiprocessing.Queue()

        for _ in range(self.num_processes):
            process = multiprocessing.Process(target=self._worker)
            self.processes.append(process)
            process.start()

        return self

    def __exit__(self, exc_type, exc_value, traceback):
        for process in self.processes:
            # Sending a sentinel value to signal worker processes to exit
            self.queue.put(None)
        for process in self.processes:
            process.join()

    def _worker(self):
        while True:
            number = self.queue.get()
            if number is None:
                break
            calculate_square(number)

def calculate_square(number):
    result = number * number
    print(f"The square of {number} is {result}")

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]

    # Usage
    with ProcessPool(3) as pool:
        for num in numbers:
            pool.queue.put(num)

    # Processes are automatically started and
     # joined when exiting the 'with' block

ProcessPool 上下文管理器通過在一組工作進程之間分配計算（整數平方的計算）以進行同步處理，從而有效地管理一組工作進程。通過利用這種形式的並行性，可以更明智地分配可用的中央處理單元，並且與在單個進程中順序執行操作相比，可以加快任務完成速度。

使用基於函數的方法的上下文管理器

contextlib 模塊提供了 @contextmanager 裝飾器，它有助於通過生成器函數創建上下文管理器。裝飾器允許通過添加新功能來增強函數的行為，而不改變其原始形式。

在基於裝飾器的方法的上下文中，利用生成器函數允許使用“yield”和“finally”語句，以便分別劃分資源的獲取和釋放。

 from contextlib import contextmanager

@contextmanager
def custom_context():
    # Code to acquire the resource
    resource = ...

    try:
        yield resource # Resource is provided to the with block
    finally:
        # Code to release the resource
        pass

為了創建用於測量給定代碼塊的執行時間的上下文管理器，可以利用面向功能的方法。這涉及在指定的代碼塊中合併計時器機制，並隨後在完成時計算其執行的持續時間。

 import time
from contextlib import contextmanager

@contextmanager
def timing_context():
    start_time = time.time()

    try:
        yield
    finally:
        end_time = time.time()
        elapsed_time = end_time - start_time
        print(f"Elapsed time: {elapsed_time} seconds")

# Usage
with timing_context():
    # Code block to measure execution time
    time.sleep(2)

時機\_上下文

使用任一方法，都可以創建定制的上下文管理器，其中包含複雜的資源管理細節和重複任務，從而優化代碼組織並促進維護工作。

嵌套上下文管理器

在需要同時管理多個資源的情況下，嵌套上下文管理器可能很有優勢。通過使用嵌套上下文，人們可以維持不間斷、無故障的操作序列，同時根據需要勤奮地獲取和釋放每個資源。

在某些情況下，程序可能需要從文件中檢索信息並將其存儲在數據庫中。處理這兩種不同的資源提出了一項挑戰，可以通過使用彼此嵌套的上下文管理器來解決。

 import sqlite3

class DatabaseConnection:
    def __enter__(self):
        self.connection = sqlite3.connect('lite.db')
        return self.connection

    def __exit__(self, exc_type, exc_value, traceback):
        self.connection.close()

# Using nested context managers
with DatabaseConnection() as db_conn, open('data.txt', 'r') as file:
    cursor = db_conn.cursor()

    # Create the table if it doesn't exist
    cursor.execute("CREATE TABLE IF NOT EXISTS data_table (data TEXT)")

    # Read data from file and insert into the database
    for line in file:
        data = line.strip()
        cursor.execute("INSERT INTO data_table (data) VALUES (?)", (data,))

    db_conn.commit()

在這種情況下，“DatabaseConnection”上下文管理器負責管理數據庫連接，而固有的“open()”上下文管理器則負責處理文件。

為了有效地管理文件和數據庫連接，將這兩個上下文包含在單個語句中非常重要。這可以確保在文件讀取或數據庫插入過程中出現任何異常時，兩種資源都會得到正確釋放。

使用裝飾器自定義函數

有效的資源管理是軟件系統實現最佳性能和穩定性的基本先決條件。資源洩漏的發生可能會導致內存積累過多，從而導致效率降低和潛在的漏洞。事實證明，利用上下文管理器提供了一種解決與資源管理相關的挑戰的改進方法。