Python MySQL连接池如何高效管理与优化？-杰瑞科技汇

Python MySQL 数据库连接池

在 Python 中使用 MySQL 数据库时，使用连接池可以显著提高性能，避免频繁创建和销毁连接的开销，以下是几种常用的 MySQL 连接池实现方式：

（图片来源网络，侵删）

使用 `mysql-connector-python` 自带连接池

import mysql.connector
from mysql.connector import pooling
# 创建连接池
dbconfig = {
    "host": "localhost",
    "user": "your_username",
    "password": "your_password",
    "database": "your_database"
}
connection_pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=5,
    **dbconfig
)
# 从连接池获取连接
connection = connection_pool.get_connection()
cursor = connection.cursor()
# 执行查询
cursor.execute("SELECT * FROM your_table")
result = cursor.fetchall()
# 关闭游标和连接
cursor.close()
connection.close()  # 将连接返回到池中

使用 `DBUtils` 连接池

DBUtils 是一个通用的 Python 数据库连接池库,支持多种数据库。

from DBUtils.PooledDB import PooledDB
import pymysql
# 创建连接池
pool = PooledDB(
    creator=pymysql,  # 使用 pymysql 作为连接创建者
    maxconnections=6,  # 连接池最大连接数
    mincached=2,      # 初始化时，链接池中至少创建的空闲的链接
    maxcached=4,      # 链接池中最多闲置的链接
    host='localhost',
    user='your_username',
    password='your_password',
    database='your_database',
    port=3306
)
# 从连接池获取连接
connection = pool.connection()
cursor = connection.cursor()
# 执行查询
cursor.execute("SELECT * FROM your_table")
result = cursor.fetchall()
# 关闭游标和连接
cursor.close()
connection.close()  # 将连接返回到池中

使用 `PyMySQL` + `DBUtils` (推荐组合)

这是目前比较流行的组合方式：

from DBUtils.PooledDB import PooledDB
import pymysql
# 创建连接池
pool = PooledDB(
    creator=pymysql,
    maxconnections=10,
    mincached=2,
    maxcached=5,
    host='localhost',
    user='your_username',
    password='your_password',
    database='your_database',
    port=3306,
    charset='utf8mb4'
)
# 使用连接池
def query(sql, params=None):
    conn = None
    cursor = None
    try:
        conn = pool.connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)  # 返回字典类型结果
        cursor.execute(sql, params or ())
        result = cursor.fetchall()
        return result
    except Exception as e:
        print(f"Database error: {e}")
        if conn:
            conn.rollback()
        raise
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()
# 使用示例
users = query("SELECT * FROM users WHERE age > %s", (18,))
print(users)

使用 `SQLAlchemy` 连接池

SQLAlchemy 也内置了连接池功能：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 创建引擎，自动管理连接池
engine = create_engine(
    'mysql+pymysql://your_username:your_password@localhost/your_database',
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600
)
# 创建会话工厂
Session = sessionmaker(bind=engine)
# 使用会话
session = Session()
try:
    users = session.execute("SELECT * FROM users WHERE age > :age", {"age": 18}).fetchall()
    print(users)
finally:
    session.close()