python中单例模式应用

数据库连接池单例模式
 

1. 为什么使用单例模式

创建数据库连接是一个昂贵的过程（涉及网络通信、认证等）。单例模式的连接池可以在程序启动时初始化一组连接，并在整个生命周期中重用这些连接，而不是每次请求都新建连接。同时还可以控制连接数量，防止资源耗尽。

2.代码

下面这段代码实现了一个数据库连接池，并且通过单例模式确保整个程序中只有一个连接池实例，包含扩容机制。避免频繁创建和销毁数据库连接，通过单例复用连接池：

通过单例模式确保全局只有一个连接池实例。
使用线程锁和条件变量实现线程安全，处理多线程环境下的并发访问。
提供连接的创建、获取、归还和关闭功能。
通过多线程测试验证连接池的并发性能。

import pymysql
import threading
from pymysql import Error


class DatabasePool:
    _instance = None

    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
            # 数据库配置
            cls._instance.config = {
                'host': 'localhost',
                'user': 'root',
                'password': '111111',
                'database': 'test',
                'charset': 'utf8mb4',
                'cursorclass': pymysql.cursors.DictCursor
            }
            # 使用可重入锁和条件变量
            cls._instance.lock = threading.RLock()
            cls._instance.condition = threading.Condition(cls._instance.lock)
            cls._instance.max_connections = 10
            cls._instance.connections = []  # 未使用的连接
            cls._instance.in_use = set()   # 正在使用的连接
            cls._instance.init_pool()
        return cls._instance    # 返回唯一的连接池实例

    def init_pool(self):
        """初始化连接池（线程安全）"""
        with self.lock:
            for _ in range(self.max_connections):
                self.add_connection()

    def create_connection(self):
        """创建单个数据库连接（无需加锁）"""
        try:
            conn = pymysql.connect(**self.config)
            print(f"成功创建连接：{conn._sock.getsockname()}")
            return conn
        except Error as e:
            print(f"连接创建失败: {e}")
            return None

    def add_connection(self):
        """向连接池添加连接（线程安全）"""
        if len(self.connections) + len(self.in_use) < self.max_connections:
            conn = self.create_connection()
            if conn:
                self.connections.append(conn)

    def get_connection(self):
        """获取连接（线程安全）"""
        with self.condition:
            while not self.connections:
                if len(self.in_use) < self.max_connections:
                    self.add_connection()
                else:
                    print("连接池已满，等待连接归还...")
                    self.condition.wait()  # 等待连接归还
            conn = self.connections.pop()
            self.in_use.add(conn)
            print(f"获取连接：{conn._sock.getsockname()}")
            return conn

    def release_connection(self, conn):
        """归还连接（线程安全）"""
        with self.condition:
            if conn in self.in_use:
                self.in_use.remove(conn)
                if conn.open:
                    self.connections.append(conn)
                    print(f"归还连接：{conn._sock.getsockname()}")
                    self.condition.notify()  # 通知等待的线程
                else:
                    print("警告：连接已关闭，直接丢弃")

    def close_pool(self):
        """关闭所有连接（线程安全）"""
        with self.lock:
            for conn in self.connections + list(self.in_use):
                if conn.open:
                    conn.close()
            self.connections.clear()
            self.in_use.clear()
            print("所有数据库连接已关闭")

# 多线程测试示例
if __name__ == "__main__":
    import concurrent.futures

    def worker(thread_id):
        pool = DatabasePool()
        conn = pool.get_connection()
        try:
            with conn.cursor() as cursor:
                cursor.execute("SELECT SLEEP(1)")  # 模拟耗时操作
                print(f"线程 {thread_id} 执行查询")
        finally:
            pool.release_connection(conn)

    # 创建连接池
    pool = DatabasePool()

    # 使用15个线程并发测试，超过最大连接数
    with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
        futures = [executor.submit(worker, i) for i in range(15)]
        for future in concurrent.futures.as_completed(futures):
            future.result()

    pool.close_pool()

3. 核心：__new__ 方法

class DatabasePool:
    _instance = None  # 类变量，用于保存唯一的实例

    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
            # 数据库配置
            cls._instance.config = {
                'host': 'localhost',
                'user': 'root',
                'password': '111111',
                'database': 'test',
                'charset': 'utf8mb4',
                'cursorclass': pymysql.cursors.DictCursor
            }
            # 使用可重入锁和条件变量
            cls._instance.lock = threading.RLock()
            cls._instance.condition = threading.Condition(cls._instance.lock)
            cls._instance.max_connections = 10
            cls._instance.connections = []  # 未使用的连接
            cls._instance.in_use = set()   # 正在使用的连接
            cls._instance.init_pool()
        return cls._instance    # 返回唯一的连接池实例

• ​作用：确保无论创建多少次 DatabasePool()，都只会生成同一个实例。
• ​示例：

  pool1 = DatabasePool()  # 第一次创建，初始化连接池
  pool2 = DatabasePool()  # 直接返回 pool1 的实例
  print(pool1 is pool2)   # 输出 True

3. 总结

我们可以把连接池想象成一个共享铅笔盒（全班共享）：

• ​锁：管理员（锁）确保一次只有一个人能拿铅笔。
• ​连接：铅笔盒里的铅笔（初始10支）。
• ​动态扩容：当铅笔用完时，管理员临时制作新铅笔。
• ​归还机制：用完后必须归还，否则其他人无法使用。