Python 資料庫連線池教學

簡介

在開發 Web API、背景服務或任何需要大量存取資料庫的應用程式時，資料庫連線的建立與關閉往往是效能瓶頸。每一次 connect() 都會觸發網路握手、認證、資源分配等步驟，若在每個請求裡都重新建立連線，系統的吞吐量會急速下降，甚至因為同時開啟過多連線而被資料庫端斷線。

為了解決這個問題，連線池（Connection Pool） 應運而生。它會在程式啟動時預先建立一定數量的連線，然後在程式需要時直接從池中取出，使用完畢再歸還，極大降低了連線建立的開銷，同時也能控制同時佔用的連線數量，避免過度消耗資料庫資源。

本篇文章將從概念、實作、常見陷阱與最佳實踐，帶你一步步了解在 Python 中如何使用連線池，並提供多個實用範例，讓你能立即在專案中套用。

---

核心概念

1. 連線池的工作原理

1. 初始化：程式啟動時依設定建立 N 個資料庫連線，放入池中。
2. 取得連線：當程式需要執行 SQL 時，向池請求一個可用連線。
3. 使用連線：執行完查詢或更新後，不要 手動關閉連線，而是 歸還 給池。
4. 回收與重建：若連線失效（例如網路斷線），池會自動回收並重新建立新連線。

重點：連線池的目標是 重用 而非 持續新增，所以 歸還 是最關鍵的步驟。

2. 常見的 Python 連線池實作方式

套件	支援的資料庫	特色
psycopg2.pool	PostgreSQL	輕量、內建於 psycopg2，適合直接操作
mysql.connector.pooling	MySQL / MariaDB	官方提供的 pool，設定簡單
SQLAlchemy	多種（PostgreSQL、MySQL、SQLite…）	高階 ORM，內建多種 pool（QueuePool、NullPool）
DBUtils	多種	針對 DB‑API 2.0 的通用連線池，提供 Thread‑Safe、Persistent 等模式

以下將以 psycopg2、mysql‑connector‑python、SQLAlchemy 為例，示範如何在不同情境下建立與使用連線池。

3. 程式碼範例

3.1 使用 psycopg2 的 SimpleConnectionPool（PostgreSQL）

import psycopg2
from psycopg2 import pool

# 建立一個最小 2、最大 10 的連線池
pg_pool = pool.SimpleConnectionPool(
    minconn=2,
    maxconn=10,
    user="postgres",
    password="your_password",
    host="127.0.0.1",
    port="5432",
    database="testdb"
)

def query_user(user_id: int):
    # 從池中取得連線
    conn = pg_pool.getconn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM users WHERE id = %s;", (user_id,))
            result = cur.fetchone()
            return result
    finally:
        # 使用完畢一定要把連線歸還
        pg_pool.putconn(conn)

# 測試
if __name__ == "__main__":
    print(query_user(1))

說明

• SimpleConnectionPool 為最簡易的實作，只需要 getconn() / putconn()。
• 若發生例外，finally 區塊確保連線不會遺失，避免池子耗盡。

---

3.2 使用 mysql‑connector‑python 的 pooling（MySQL）

import mysql.connector
from mysql.connector import pooling

# 建立連線池，設定 pool_name 與 pool_size
cnxpool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=5,
    host="localhost",
    user="root",
    password="your_password",
    database="testdb"
)

def insert_log(message: str):
    # 直接呼叫 get_connection() 取得連線
    cnx = cnxpool.get_connection()
    try:
        cursor = cnx.cursor()
        cursor.execute(
            "INSERT INTO logs (msg, created_at) VALUES (%s, NOW())",
            (message,)
        )
        cnx.commit()
    finally:
        # close() 會自動把連線歸還給 pool
        cnx.close()

if __name__ == "__main__":
    insert_log("連線池測試訊息")

說明

• MySQLConnectionPool 內部已實作 Thread‑Safe，多執行緒環境下也能安全使用。
• 呼叫 cnx.close() 時，實際上是把連線回收到池子，而不是關閉底層 socket。

---

3.3 使用 SQLAlchemy 的 QueuePool（適用多種資料庫）

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

# 建立 Engine，預設使用 QueuePool
engine = create_engine(
    "postgresql+psycopg2://postgres:your_password@localhost:5432/testdb",
    pool_size=10,          # 最大連線數
    max_overflow=5,        # 超過 pool_size 時可臨時建立的連線數
    pool_timeout=30,       # 等待可用連線的最長秒數
    pool_pre_ping=True    # 每次取出前先 ping，避免失效連線
)

# 建立 SessionFactory
Session = sessionmaker(bind=engine)

def get_all_products():
    session = Session()
    try:
        result = session.execute(text("SELECT * FROM products"))
        return result.fetchall()
    finally:
        session.close()   # close() 會把連線歸還給 pool

if __name__ == "__main__":
    products = get_all_products()
    for p in products:
        print(p)

說明

• create_engine 內部即完成連線池的設定，不需要額外呼叫 取得/歸還連線。
• pool_pre_ping=True 能在連線失效時自動重新建立，降低因斷線產生的例外。

---

3.4 使用 DBUtils 的 PooledDB（通用 DB‑API）

from dbutils.pooled_db import PooledDB
import sqlite3   # 以 SQLite 為例（實際上 SQLite 本身不支援多連線）

# SQLite 只做示範，實務上多用於 PostgreSQL/MySQL
pool = PooledDB(
    creator=sqlite3,          # 任何符合 DB‑API 2.0 的模組
    maxconnections=5,
    timeout=20,
    database="example.db"
)

def fetch_settings():
    conn = pool.connection()
    try:
        cur = conn.cursor()
        cur.execute("SELECT key, value FROM settings")
        return dict(cur.fetchall())
    finally:
        conn.close()   # 歸還給 pool

if __name__ == "__main__":
    print(fetch_settings())

說明

• PooledDB 為 通用型，只要提供 creator（符合 DB‑API）即可使用。
• 這對於需要在同一套程式中支援多種資料庫的專案非常有幫助。

---

3.5 手動實作簡易連線池（學習示例）

import threading
import queue
import psycopg2

class SimplePool:
    def __init__(self, minsize, maxsize, **conn_kwargs):
        self._pool = queue.Queue(maxsize)
        self._maxsize = maxsize
        for _ in range(minsize):
            self._pool.put(self._create_conn(**conn_kwargs))
        self._conn_kwargs = conn_kwargs
        self._lock = threading.Lock()

    def _create_conn(self, **kw):
        return psycopg2.connect(**kw)

    def get(self):
        try:
            return self._pool.get_nowait()
        except queue.Empty:
            with self._lock:
                # 若已達上限則等候
                if self._pool.qsize() + 1 > self._maxsize:
                    return self._pool.get()
                return self._create_conn(**self._conn_kwargs)

    def put(self, conn):
        self._pool.put(conn)

# 使用範例
pool = SimplePool(2, 5, user="postgres", password="pw", host="localhost", database="testdb")

def count_rows():
    conn = pool.get()
    try:
        cur = conn.cursor()
        cur.execute("SELECT COUNT(*) FROM orders")
        return cur.fetchone()[0]
    finally:
        pool.put(conn)

print(count_rows())

說明

• 這段程式碼展示了 Queue + threading.Lock 的組合，適合作為學習連線池概念的練習。
• 實務上建議直接使用成熟套件，以避免隱藏的競爭條件與資源洩漏。

---

常見陷阱與最佳實踐

陷阱	說明	解決方式
忘記歸還連線	取出連線後直接 close()（或不處理）會導致池子耗盡，最終拋出 PoolError: No more connections。	使用 try/finally、with 語法或 ORM 的 session.close()，確保每條路徑都有歸還。
過大 pool_size	設定過高的最大連線數會讓資料庫端的連線上限被突破，導致拒絕服務。	依照資料庫的 max_connections 與應用的併發需求，適度調整 pool_size 與 max_overflow。
連線失效未偵測	長時間閒置的連線可能被防火牆或資料庫端回收，使用時會拋出 OperationalError。	開啟 pool_pre_ping=True（SQLAlchemy）或在取得連線前自行 ping，讓池子自動重建失效連線。
在多執行緒/多程序環境中共享同一個 Engine	若不使用 Thread‑Safe 的 pool，可能發生競爭條件或資料損毀。	使用套件自帶的 Thread‑Safe pool（如 psycopg2.pool、MySQLConnectionPool），或在每個執行緒/程序內部建立獨立的 Engine。
在長時間交易（transaction）中持有連線	大量寫入或長時間鎖定會佔用池子資源，其他請求被迫等待。	盡量縮短 transaction，只在需要時才開啟，完成後立即提交或回滾並歸還。

最佳實踐清單

1. 統一管理連線池：將 pool / engine 的建立放在單例或全域模組中，避免重複建立。
2. 使用 context manager：with pool.getconn() as conn: 或 with Session() as s: 能自動處理歸還。
3. 設定合理的 max_overflow：讓突發流量有彈性，同時不會無限制擴張。
4. 監控連線使用率：透過 pool.checkedout()（SQLAlchemy）或自行統計，了解是否需要調整大小。
5. 在測試環境模擬高併發：使用 locust、ab 等工具測試連線池在壓力下的行為，避免上線後出現瓶頸。

---

實際應用場景

場景	為何需要連線池	推薦套件
Web 框架（Flask / Django）	每個 HTTP 請求都會執行多次 DB 查詢，請求量可能達千級/秒。	Django 預設使用 django.db.backends 的 pool，Flask 可結合 SQLAlchemy 或 psycopg2.pool。
背景工作者（Celery、RQ）	任務可能同時跑上百個，若每個任務都自行建立連線會造成資源浪費。	Celery 可在 worker 啟動時建立全域 engine，使用 QueuePool。
資料分析腳本	大量讀取資料庫後寫入檔案或進行統計，連續執行多次查詢。	DBUtils 的 PooledDB，簡單且支援多種 DB‑API。
微服務間的資料共享	多個微服務同時存取同一資料庫，需要控制總連線數以免衝突。	SQLAlchemy + pgbouncer（PostgreSQL）共同使用，pgbouncer 本身即為外部連線池。
長線交易（金融、訂單）	需要在同一筆交易中執行多個步驟，且必須確保原子性。	使用 Transaction 與 Session，配合 pool_size 與 max_overflow 讓交易期間不被其他請求阻塞。

---

總結

• 資料庫連線池 是提升 Python 應用程式效能與穩定性的關鍵工具，能顯著減少連線建立的開銷，並避免因過度佔用連線而導致的資源枯竭。
• Python 生態提供了 psycopg2、mysql‑connector‑python、SQLAlchemy、DBUtils 等成熟套件，根據資料庫類型與專案需求選擇最合適的實作即可。
• 實作時務必 確保每一次取得的連線都能正確歸還，並根據資料庫的最大連線數、預期併發量設定合理的 pool_size、max_overflow。
• 透過 pre‑ping、監控與適當的例外處理，可以避免因連線失效產生的不可預期錯誤。
• 最後，在測試階段就導入連線池，並使用壓力測試驗證其行為，才能在正式上線時確保系統的高可用與高效能。

掌握連線池的使用，等於為你的 Python 應用打下了穩固的基礎，未來再面對更大規模的流量時，也能輕鬆應對。