Python中如何仅初始化一次数据库连接并在运行时复用？现有实现是否正确？

现有实现的评估 #

• 连接复用逻辑正确性：你当前采用的单例设计中，__new__方法仅在第一次实例化Database类时执行数据库连接创建逻辑，后续所有实例化操作都会返回同一个全局实例，理论上数据库连接确实只会被初始化一次，满足连接复用的核心需求。
• 存在的问题：
  • 未实现上下文管理器协议：当前类缺少__enter__和__exit__方法，无法直接用with语句调用，运行你给出的foo函数会直接报错。
  • 全局共用单个游标存在风险：所有操作共用同一个cursor实例，多线程并发操作时会出现游标资源竞争、查询结果串扰的问题，也会导致事务处理逻辑混乱。
  • 无断连重连机制：如果程序运行过程中数据库连接超时断开，现有实现不会自动重连，会直接抛出操作异常。

改进后的实现代码 #

import mysql.connector
from mysql.connector import Error

class Database(object):
    _instance = None
    _connection = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = object.__new__(cls)
            cls._instance._connect()
        return cls._instance
    
    def _connect(self):
        """内部方法：创建数据库连接"""
        try:
            self._connection = mysql.connector.connect(
                host="127.0.0.1", 
                user="root", 
                password="", 
                database="db_test"
            )
            print("Connection established")
        except Error as error:
            print(f"Error: Connection not established {error}")
            self._connection = None
    
    def __enter__(self):
        """上下文管理器进入：新建游标返回"""
        # 检测连接是否有效，断开则自动重连
        if not self._connection or not self._connection.is_connected():
            self._connect()
        # 每次上下文新建独立游标，避免资源竞争
        self.cursor = self._connection.cursor()
        return self.cursor
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """上下文管理器退出：处理事务、关闭游标"""
        if exc_type is None:
            # 无异常则提交事务
            self._connection.commit()
        else:
            # 有异常则回滚
            self._connection.rollback()
            print(f"Database operation error: {exc_val}")
        # 关闭当前游标
        self.cursor.close()
        self.cursor = None

改进点说明 #

• 保持单例特性，全局仅创建一个数据库连接，避免频繁新建连接的开销
• 实现了完整的上下文管理器协议，兼容你原有的with Database() as cursor调用方式
• 每次进入上下文时新建独立游标，退出时自动关闭游标，避免多线程操作的资源冲突
• 新增连接有效性检测和自动重连逻辑，解决连接超时断开的问题
• 新增事务自动提交/回滚逻辑，避免手动管理事务的遗漏问题

可选优化：高并发场景用连接池 #

如果你的项目并发量较高，单连接无法满足性能需求，可以替换为mysql官方的连接池实现，连接池会预先创建多个可用连接，请求时分配空闲连接，用完归还，比单连接的并发支撑能力更强：

from mysql.connector import pooling

db_pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=5,
    host="127.0.0.1", 
    user="root", 
    password="", 
    database="db_test"
)

# 使用时从连接池获取连接
def foo():
    conn = db_pool.get_connection()
    with conn.cursor() as cursor:
        cursor.execute("STATEMENT")
    conn.commit()
    conn.close() # 此处的close是把连接归还连接池，不是真正关闭

内容的提问来源于stack exchange，提问作者Maik Hasler