### 在 Python 中使用 PooledDB 连接数据库并捕获相关错误的最佳实践 为了确保在使用 `PooledDB` 时能够正确地捕获和处理错误，以下是最佳实践的具体实现方法。这种方法不仅提高了代码的健壮性，还减少了因异常而导致的服务中断风险。 #### 创建连接池 首先需要导入必要的模块，并配置连接池的相关参数。这里以 MySQL 数据库为例： ```python from DBUtils.PooledDB import PooledDB import pymysql # 配置连接池 pool = PooledDB( creator=pymysql, # 使用 pymysql 作为数据库驱动 maxconnections=10, # 最大连接数 mincached=2, # 初始化时缓存的最小连接数 maxcached=5, # 缓存的最大连接数 maxshared=3, # 共享的最大连接数 blocking=True, # 当连接数达到上限时是否阻塞等待 maxusage=None, # 单个连接最大复用次数 setsession=[], # 设置会话前执行的命令列表 ping=0, # 检查数据库连接状态的时间间隔（单位：秒） host='localhost', port=3306, user='root', password='password', database='test_db', charset='utf8' ) ``` 以上代码定义了一个具有固定属性的连接池实例，这些属性决定了连接池的行为特性[^1]。 --- #### 获取连接与执行 SQL 查询 通过连接池获取数据库连接后，可以执行 SQL 查询或更新操作。在此过程中应始终使用 `try...except...finally` 结构来捕获可能发生的异常，并确保资源被正确释放。 ```python def execute_query(query): connection = None cursor = None try: # 从连接池中获取连接 connection = pool.connection() # 创建游标对象 cursor = connection.cursor() # 执行查询 cursor.execute(query) # 提交事务（如果有修改操作的话） if query.strip().upper().startswith(('INSERT', 'UPDATE', 'DELETE')): connection.commit() # 返回查询结果 return cursor.fetchall() except pymysql.MySQLError as db_error: print(f"Database error occurred: {db_error}") if connection: connection.rollback() # 发生错误时回滚事务 finally: # 关闭游标和连接 if cursor: cursor.close() if connection: connection.close() ``` 上述函数封装了从连接池获取连接、执行 SQL 查询以及处理异常的过程。它能够在遇到数据库错误时打印详细的错误信息，并且回滚未完成的事务以防数据不一致[^2]。 --- #### 处理常见错误 在实际开发中可能会遇到一些常见的数据库错误，例如连接丢失 (`OperationalError`) 或者超长字符串插入失败等。针对这些问题，可以通过增强异常处理逻辑进一步提升系统的稳定性。 ##### 示例：处理 `(2006, 'MySQL server has gone away')` 当长时间保持同一个数据库连接而没有活动时，服务器可能会主动断开连接。为了避免这种情况影响程序正常运行，可以在每次请求之前先验证连接的有效性。 ```python def validate_and_execute(query): connection = None cursor = None while True: try: # 从连接池中获取连接 connection = pool.connection() # 创建游标对象 cursor = connection.cursor() # 验证连接有效性 cursor.execute("SELECT 1") # 简单测试查询 # 正常情况下继续执行目标查询 break except pymysql.OperationalError as op_error: print(f"Connection lost, retrying: {op_error}") continue # 如果连接失效，则重新尝试获取新连接 try: # 执行真正的业务查询 cursor.execute(query) return cursor.fetchall() except Exception as ex: print(f"An unexpected error occurred: {ex}") raise finally: if cursor: cursor.close() if connection: connection.close() ``` 这段代码实现了对无效连接的检测机制，一旦发现连接不可用就立即丢弃旧连接并重新建立新的连接[^1]。 --- #### 总结 通过合理运用 `PooledDB` 和完善的异常处理策略，不仅可以有效减少频繁创建销毁数据库连接带来的性能损耗，还能显著提高应用程序面对突发状况时的恢复能力。务必注意的是，在多线程环境下要特别小心管理好各个线程间的资源共享问题以免引发竞态条件或其他同步类问题。 ---