python中单例模式应用

数据库毗连池单例模式
 

1. 为什么使用单例模式

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

2.代码

下面这段代码实现了一个数据库毗连池，而且通过单例模式确保整个步调中只有一个毗连池实例，包罗扩容机制。克制频仍创建和烧毁数据库毗连，通过单例复用毗连池：
通过单例模式确保全局只有一个毗连池实例。
使用线程锁和条件变量实现线程安全，处置处罚多线程环境下的并发访问。
提供毗连的创建、获取、归还和关闭功能。
通过多线程测试验证毗连池的并发性能。

2. import pymysql
3. import threading
4. from pymysql import Error
6. class DatabasePool:
7.     _instance = None
9.     def __new__(cls):
10.         if not cls._instance:
11.             cls._instance = super().__new__(cls)
12.             # 数据库配置
13.             cls._instance.config = {
14.                 'host': 'localhost',
15.                 'user': 'root',
16.                 'password': '111111',
17.                 'database': 'test',
18.                 'charset': 'utf8mb4',
19.                 'cursorclass': pymysql.cursors.DictCursor
20.             }
21.             # 使用可重入锁和条件变量
22.             cls._instance.lock = threading.RLock()
23.             cls._instance.condition = threading.Condition(cls._instance.lock)
24.             cls._instance.max_connections = 10
25.             cls._instance.connections = []  # 未使用的连接
26.             cls._instance.in_use = set()   # 正在使用的连接
27.             cls._instance.init_pool()
28.         return cls._instance    # 返回唯一的连接池实例
30.     def init_pool(self):
31.         """初始化连接池（线程安全）"""
32.         with self.lock:
33.             for _ in range(self.max_connections):
34.                 self.add_connection()
36.     def create_connection(self):
37.         """创建单个数据库连接（无需加锁）"""
38.         try:
39.             conn = pymysql.connect(**self.config)
40.             print(f"成功创建连接：{conn._sock.getsockname()}")
41.             return conn
42.         except Error as e:
43.             print(f"连接创建失败: {e}")
44.             return None
46.     def add_connection(self):
47.         """向连接池添加连接（线程安全）"""
48.         if len(self.connections) + len(self.in_use) < self.max_connections:
49.             conn = self.create_connection()
50.             if conn:
51.                 self.connections.append(conn)
53.     def get_connection(self):
54.         """获取连接（线程安全）"""
55.         with self.condition:
56.             while not self.connections:
57.                 if len(self.in_use) < self.max_connections:
58.                     self.add_connection()
59.                 else:
60.                     print("连接池已满，等待连接归还...")
61.                     self.condition.wait()  # 等待连接归还
62.             conn = self.connections.pop()
63.             self.in_use.add(conn)
64.             print(f"获取连接：{conn._sock.getsockname()}")
65.             return conn
67.     def release_connection(self, conn):
68.         """归还连接（线程安全）"""
69.         with self.condition:
70.             if conn in self.in_use:
71.                 self.in_use.remove(conn)
72.                 if conn.open:
73.                     self.connections.append(conn)
74.                     print(f"归还连接：{conn._sock.getsockname()}")
75.                     self.condition.notify()  # 通知等待的线程
76.                 else:
77.                     print("警告：连接已关闭，直接丢弃")
79.     def close_pool(self):
80.         """关闭所有连接（线程安全）"""
81.         with self.lock:
82.             for conn in self.connections + list(self.in_use):
83.                 if conn.open:
84.                     conn.close()
85.             self.connections.clear()
86.             self.in_use.clear()
87.             print("所有数据库连接已关闭")
89. # 多线程测试示例
90. if __name__ == "__main__":
91.     import concurrent.futures
93.     def worker(thread_id):
94.         pool = DatabasePool()
95.         conn = pool.get_connection()
96.         try:
97.             with conn.cursor() as cursor:
98.                 cursor.execute("SELECT SLEEP(1)")  # 模拟耗时操作
99.                 print(f"线程 {thread_id} 执行查询")
100.         finally:
101.             pool.release_connection(conn)
103.     # 创建连接池
104.     pool = DatabasePool()
106.     # 使用15个线程并发测试，超过最大连接数
107.     with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
108.         futures = [executor.submit(worker, i) for i in range(15)]
109.         for future in concurrent.futures.as_completed(futures):
110.             future.result()
112.     pool.close_pool()

复制代码

3. 核心：__new__ 方法

2. class DatabasePool:
3.     _instance = None  # 类变量，用于保存唯一的实例
5.     def __new__(cls):
6.         if not cls._instance:
7.             cls._instance = super().__new__(cls)
8.             # 数据库配置
9.             cls._instance.config = {
10.                 'host': 'localhost',
11.                 'user': 'root',
12.                 'password': '111111',
13.                 'database': 'test',
14.                 'charset': 'utf8mb4',
15.                 'cursorclass': pymysql.cursors.DictCursor
16.             }
17.             # 使用可重入锁和条件变量
18.             cls._instance.lock = threading.RLock()
19.             cls._instance.condition = threading.Condition(cls._instance.lock)
20.             cls._instance.max_connections = 10
21.             cls._instance.connections = []  # 未使用的连接
22.             cls._instance.in_use = set()   # 正在使用的连接
23.             cls._instance.init_pool()
24.         return cls._instance    # 返回唯一的连接池实例

复制代码

• ​作用：确保无论创建多少次 DatabasePool()，都只会天生同一个实例。
  
• ​示例：
  2. pool1 = DatabasePool()  # 第一次创建，初始化连接池
  3. pool2 = DatabasePool()  # 直接返回 pool1 的实例
  4. print(pool1 is pool2)   # 输出 True

  复制代码

3. 总结

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

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