Python数据库连接池教程 - DBUtils连接池配置与性能优化

# 安装DBUtils（2.x版本）
pip install DBUtils

# 注意：DBUtils 2.x 仅支持Python 3.6+

from dbutils.pooled_db import PooledDB
import pymysql

# 创建连接池
pool = PooledDB(
    creator=pymysql,           # 数据库驱动模块
    maxconnections=10,         # 连接池允许的最大连接数
    mincached=2,               # 初始化时创建的最小空闲连接数
    maxcached=5,               # 连接池最大空闲连接数
    maxshared=3,               # 最大共享连接数（0或None表示全部共享）
    blocking=True,             # 连接池满时是否阻塞等待（True等待，False抛出异常）
    maxusage=None,             # 连接最大复用次数（None表示无限制）
    setsession=[],             # 执行前的会话设置命令列表
    ping=1,                    # 检查连接是否可用（0=不检查, 1=默认, 2=使用时检查, 4=获取时检查）
    # 以下是传递给pymysql.connect()的参数
    host='localhost',
    port=3306,
    user='root',
    password='your_password',
    database='test_db',
    charset='utf8mb4'
)

# 从连接池获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute('SELECT VERSION()')
result = cursor.fetchone()
print(f"MySQL版本：{result}")

# 关闭连接（实际上是归还到连接池）
cursor.close()
conn.close()

from dbutils.persistent_db import PersistentDB
import pymysql

# 创建线程专用连接
persistent_db = PersistentDB(
    creator=pymysql,
    maxusage=1000,             # 每个连接最大使用次数
    # pymysql连接参数
    host='localhost',
    port=3306,
    user='root',
    password='your_password',
    database='test_db',
    charset='utf8mb4'
)

# 获取连接（同一线程获取到的是同一个连接）
conn = persistent_db.connection()
cursor = conn.cursor()
cursor.execute('SELECT 1')
print(cursor.fetchone())

cursor.close()
conn.close()

"""
数据库连接池完整示例 - 基于DBUtils + PyMySQL
"""
from dbutils.pooled_db import PooledDB
import pymysql
import threading
import time
from contextlib import contextmanager

class DatabasePool:
    """数据库连接池管理类"""
    
    _instance = None
    _lock = threading.Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self):
        if not hasattr(self, 'pool'):
            self.pool = PooledDB(
                creator=pymysql,
                maxconnections=10,
                mincached=2,
                maxcached=5,
                maxshared=3,
                blocking=True,
                maxusage=None,
                ping=1,
                host='localhost',
                port=3306,
                user='root',
                password='your_password',
                database='test_db',
                charset='utf8mb4',
                cursorclass=pymysql.cursors.DictCursor
            )
            print("数据库连接池初始化完成！")
    
    @contextmanager
    def get_connection(self):
        """获取数据库连接的上下文管理器"""
        conn = self.pool.connection()
        try:
            yield conn
        finally:
            conn.close()  # 归还到连接池
    
    def execute_query(self, sql, params=None):
        """执行查询并返回结果"""
        with self.get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(sql, params or ())
                return cursor.fetchall()
    
    def execute_update(self, sql, params=None):
        """执行更新操作"""
        with self.get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(sql, params or ())
                conn.commit()
                return cursor.rowcount

# 全局连接池实例
db_pool = DatabasePool()

def worker_task(worker_id):
    """模拟工作线程任务"""
    try:
        # 查询操作
        results = db_pool.execute_query('SELECT SLEEP(0.1)')
        print(f"Worker {worker_id}: 查询完成")
        
        # 更新操作
        db_pool.execute_update(
            'INSERT INTO access_log (worker_id, access_time) VALUES (%s, NOW())',
            (worker_id,)
        )
        print(f"Worker {worker_id}: 日志记录完成")
    except Exception as e:
        print(f"Worker {worker_id}: 发生错误 - {e}")

if __name__ == '__main__':
    # 创建access_log表
    db_pool.execute_update('''
        CREATE TABLE IF NOT EXISTS access_log (
            id INT AUTO_INCREMENT PRIMARY KEY,
            worker_id INT,
            access_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    # 模拟10个并发线程
    threads = []
    for i in range(10):
        t = threading.Thread(target=worker_task, args=(i,))
        threads.append(t)
        t.start()
    
    # 等待所有线程完成
    for t in threads:
        t.join()
    
    print("\n所有任务完成！")
    
    # 查看连接池统计
    results = db_pool.execute_query('SELECT COUNT(*) as count FROM access_log')
    print(f"共记录 {results[0]['count']} 次访问")

# 推荐：使用上下文管理器
with db_pool.get_connection() as conn:
    with conn.cursor() as cursor:
        cursor.execute('SELECT 1')
# 连接自动归还到连接池

# 不推荐：手动管理
conn = db_pool.pool.connection()
cursor = conn.cursor()
cursor.execute('SELECT 1')
# 如果中间抛出异常，conn.close()不会被执行，导致连接泄漏