在 Python 中使用連線池提升效能

2024-05-15碼農

來源：Python七號

在資料庫操作中，頻繁地開啟和關閉連線會消耗大量的資源和時間，在一些需要高並行的場景，我們就需要連線池技術最佳化這一問題。本文以 Python 中的 SQLite 資料庫為例，介紹如何使用連線池來提升數據操作的效率。

對於使用 SQLite 這樣的輕量級資料庫，雖然連線的開銷相對較小，但在高並行或要求快速響應的套用場景中，最佳化連線過程仍然非常重要。

import sqlite3 # Connect to database ( 這裏會很耗時 ) conn = sqlite3.connect('mydatabase.db') # Create a cursor object cursor = conn.cursor() # Create a table (if it doesn't exist) cursor.execute('''sql statement''')

首先，我們來理解一下什麽是連線池。在資料庫操作中，每次存取資料庫都需要和資料庫伺服器建立一個連線（ conn = sqlite3.connect('mydatabase.db') ），這個過程包括建立連線物件、與資料庫建立通訊、驗證等多個步驟。連線池，顧名思義，就是預建立並儲存一系列連線物件，當程式需要與資料庫互動時，直接從池中取用，用完了再放回池中，避免了頻繁建立和銷毀連線的開銷。

使用資料庫連線池的步驟：

安裝必要的 Python 庫

首先確保已經安裝了 sqlite3 模組，雖然 SQLite 在 Python 標準庫中內建了 sqlite3 介面，但是沒有實作連線池的功能，為此我們需要自行編寫連線池，或使用第三方庫，例如 sqlalchemy . 透過 pip 可以輕松安裝 :

pip install sqlalchemy

接下來分別介紹這兩種方法。

自行編寫連線池

自行編寫連線池，可以更好的理解連線池到底做了什麽。

import sqlite3 from queue import Queue classSQLiteConnectionPool: """ SQLite 連線池類，用於管理資料庫連線，避免頻繁建立和關閉連線的開銷。 """ def__init__(self, db_path, max_connections=10): """ 初始化連線池。 Args: db_path (str): 資料庫檔路徑。 max_connections (int): 連線池允許的最大連線數，預設為 10。 """ self.db_path = db_path # 資料庫檔路徑 self.max_connections = max_connections # 最大連線數 self.free_connections = Queue(maxsize=max_connections) # 儲存空閑連線的佇列 defget_connection(self): """ 從連線池獲取一個資料庫連線。 Returns: sqlite3.Connection: 一個 SQLite 資料庫連線物件。 """ if self.free_connections.empty(): # 如果佇列為空，則建立新的連線 conn = sqlite3.connect(self.db_path) else: # 否則從佇列中獲取一個空閑連線 conn = self.free_connections.get() return conn defput_connection(self, conn): """ 將一個資料庫連線放回連線池。 Args: conn (sqlite3.Connection): 要放回的資料庫連線物件。 """ self.free_connections.put(conn) # 將連線放回佇列

程式碼解釋 :

SQLiteConnectionPool 類 : 定義了連線池類，用於管理資料庫連線。

建構函式 : 接收 db_path ( 資料庫檔路徑 ) 和 max_connections ( 最大連線數 ) 參數。建立了一個 Queue 物件來儲存空閑連線，最大容量為 max_connections。

get_connection() 方法 : 嘗試從 free_connections 佇列獲取一個空閑連線。如果佇列為空，則建立一個新的連線。返回獲取到的連線。

put_connection() 方法 : 將用完的連線放回 free_connections 佇列，以便後續復用。

該連線池的使用方法：

pool = SQLiteConnectionPool('yoursqlite.db') defconcurrent_access(pool): conn = pool.get_connection() cur = conn.cursor() # 參數化查詢 cur.execute("sql") rows = cur.fetchall() pool.put_connection(conn) return rows

函式 concurrent_access 可以在高頻場景下呼叫，本質上，連線池相當於一個全域變量。

使用 sqlalchemy 建立連線池

使用 sqlalchemy 的 create_engine 函式，我們可以建立一個 SQLite 連線池。

from sqlalchemy import create_engine engine = create_engine('sqlite:///my_database.db', pool_size=10)

使用方法也比較簡單 :

from sqlalchemy import create_engine,text engine = create_engine('sqlite:///my_database.db', pool_size=10) defconcurrent_access(): with engine.connect() as conn: result = conn.execute(text("select * from reply")) for row in result: print(row)