python中单例模式应用

数据库连接池单例模式
 

1. 为什么使用单例模式

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

2.代码

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

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

import pymysql
import threading
from pymysql import Errorclass DatabasePool:_instance = Nonedef __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 = 10cls._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 connexcept Error as e:print(f"连接创建失败: {e}")return Nonedef 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 conndef 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.futuresdef 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 = 10cls._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支）。
• ​动态扩容：当铅笔用完时，管理员临时制作新铅笔。
• ​归还机制：用完后必须归还，否则其他人无法使用。