1 import time
  2 import pymysql
  3 from dbutils import persistent_db, pooled_db, simple_pooled_db, steady_db
  4 from multiprocessing.pool import ThreadPool
  5 from abc import abstractmethod,ABCMeta
  6 from queue import Queue
  7 # print(help(pooled_db.PooledDB))
  8 # pooled_db 线程安全
  9 # print(pymysql.threadsafety)
 10 
 11 db_conf = {"host": "127.0.0.1",
 12            "port": 3306,
 13            "user": "root",
 14            "password": "123456",
 15            "database": "userinfo"
 16            }
 17 
 18 
 19 class TaskConsumer(metaclass=ABCMeta):
 20     @abstractmethod
 21     def _task_worker(self, i):
 22         pass
 23 
 24     def consumer(self):
 25         task_pool = ThreadPool(self.task_pool_num)
 26         start_time = time.time()
 27         for i in range(self.task_num):
 28             task_pool.apply_async(func=self._task_worker, args=(i,))
 29         task_pool.close()
 30         task_pool.join()
 31         return time.time() - start_time
 32 
 33     def db_read(self, cur, sql, i):
 34         # time.sleep(2)
 35         cur.execute(sql)
 36         print(i, cur.fetchall())
 37 
 38 class DbPoolTest(TaskConsumer):
 39     def __init__(self, task_num, task_pool_num, max_conn, blocking=True, **db_conf):
 40         self.task_num = task_num
 41         self.task_pool_num = task_pool_num
 42         self.max_conn=max_conn
 43         self.blocking = blocking
 44         self.kwargs = db_conf
 45         self._get_db_pool()
 46 
 47     def _get_db_pool(self):
 48         self.db_pool = pooled_db.PooledDB(
 49             creator=pymysql,  # 连接对象或符合DB-API 2的数据库模块（使用链接数据库的模块）
 50             mincached=self.max_conn,  # 池中空闲连接的初始数量（默认值为0表示启动时未建立连接）
 51             maxcached=0,  # 池中最大空闲连接数（默认值0或None表示池大小不受限制）
 52             maxshared=0,
 53             # 允许的最大共享连接数（默认值为0或None表示所有连接都是专用的）；达到此最大数目后，如果已将连接请求为可共享的，则将共享它们。因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
 54             maxconnections=self.max_conn,  # 通常允许的最大连接数（默认值为0或无表示任何数量的连接）
 55             blocking=self.blocking,  # 确定超出最大值时的行为（如果将其设置为true，阻塞并等待直到连接减少，但默认情况下将报告错误）
 56             maxusage=None,  # 单个连接的最大重用次数（默认值为0或无表示无限重用）；当达到连接的最大使用次数时，该连接将自动重置（关闭并重新打开）
 57             setsession=None,  # 可用于准备会话的SQL命令的可选列表，例如 [“将日期样式设置为德语”，...]
 58             reset=True,  # 将连接返回池时应如何重置（False或None为以begin（）开始的回滚事务，出于安全考虑，默认值True始终触发回滚）
 59             failures=None,  # 如果默认（OperationalError，InternalError）不足，则应为其应用连接故障转移机制的可选异常类或异常类的元组
 60             ping=1,
 61             # 可选标志，用于控制何时使用ping（）方法检查连接（如果该方法可用）（0 = None = never, 1 = default = whenever it is requested,2 = when a cursor is created, 4 = when a query is executed, 7 = always）
 62             **self.kwargs)
 63 
 64     def _task_worker(self, i):
 65         conn = self.db_pool.connection()
 66         cur = conn.cursor()
 67         sql = "select * from username;"
 68         self.db_read(cur, sql, i)
 69         cur.close()
 70         conn.close()
 71 
 72     def close(self):
 73         self.db_pool.close()
 74 
 75 class PymysqlTest(TaskConsumer):
 76     def __init__(self,task_num, task_pool_num, **db_conf):
 77         self.task_num = task_num
 78         self.task_pool_num = task_pool_num
 79         self.kwargs = db_conf
 80 
 81     def _task_worker(self, i):
 82         conn = pymysql.Connect(**self.kwargs)
 83         cur = conn.cursor()
 84         sql = "select * from username;"
 85         self.db_read(cur, sql, i)
 86         cur.close()
 87         conn.close()
 88 
 89 
 90 class PymysqlPoolTest(TaskConsumer):
 91     def __init__(self, task_num, task_pool_num, max_conn, blocking=True, **db_conf):
 92         self.task_num = task_num
 93         self.task_pool_num = task_pool_num
 94         self.max_conn = max_conn
 95         self.blocking = blocking
 96         self.kwargs = db_conf
 97         self._get_pymysql_pool()
 98 
 99     def _get_pymysql_pool(self):
100         self.queue_pool = Queue(maxsize=self.max_conn)
101         for i in range(self.max_conn):
102             self.queue_pool.put(pymysql.Connect(**self.kwargs))
103 
104     def _task_worker(self, i):
105         conn = self.queue_pool.get()
106         cur = conn.cursor()
107         sql = "select * from username;"
108         self.db_read(cur, sql, i)
109         cur.close()
110         self.queue_pool.put(conn)
111 
112     def close(self):
113         while not self.queue_pool.empty():
114             self.queue_pool.get().close()
115 
116 if __name__ == '__main__':
117     TASK_NUM = 10000
118     TASK_POOL_NUM = 100
119 
120 
121     # 方案1：数据库连接池方案（建议设置blocking=True,maxconnections最大连接数量）
122     start_time_dbpool=time.time()
123     DB_POOL_NUM = 100 #DB_POOL_NUM>=TASK_POOL_NUM任务执行时间基本保持不变
124     db_pool_test = DbPoolTest(TASK_NUM, TASK_POOL_NUM, DB_POOL_NUM,**db_conf)
125     # start_time_dbpool = time.time()
126     time_dbpool_task = db_pool_test.consumer()
127     db_pool_test.close()
128     time_dbpool = (time.time()-start_time_dbpool, time_dbpool_task)
129 
130 
131     # 方案2：每个线程池中手动建立一个连接
132     start_time_pymysql = time.time()
133     pymysql_test = PymysqlTest(TASK_NUM, TASK_POOL_NUM, **db_conf)
134     time_pymysql_task = pymysql_test.consumer()
135     time_pymysql = (time.time()-start_time_pymysql, time_pymysql_task)
136 
137     # 方案3:手动维护数据库连接池
138     start_time_pymysqlpool = time.time()
139     DB_POOL_NUM = 100  # DB_POOL_NUM>=TASK_POOL_NUM任务执行时间基本保持不变
140     pymysql_pool_test = PymysqlPoolTest(TASK_NUM, TASK_POOL_NUM, DB_POOL_NUM, **db_conf)
141     # start_time_dbpool = time.time()
142     time_pymysqlpool_task = pymysql_pool_test.consumer()
143     pymysql_pool_test.close()
144     time_pymysqlpool = (time.time() - start_time_pymysqlpool, time_pymysqlpool_task)
145 
146 
147     print("数据库连接池方案：", time_dbpool)
148     print("线程内手动连接方案：", time_pymysql)
149     print("手动维护数据库连接池方案：", time_pymysqlpool)