数据库连接池

高并发下数据库连接使用方案：

1.使用已有数据库连接池（下文有介绍）

2.每个执行单元建立一次连接独享；（效率超低，难以维护，不建议使用）

3.手动维护线程池（本人在MAC测试目前效率最高）；

实测代码案例：

import time
import pymysql
from dbutils import persistent_db, pooled_db, simple_pooled_db, steady_db
from multiprocessing.pool import ThreadPool
from abc import abstractmethod,ABCMeta
from queue import Queue
# print(help(pooled_db.PooledDB))
# pooled_db 线程安全
# print(pymysql.threadsafety)

db_conf = {"host": "127.0.0.1",
           "port": 3306,
           "user": "root",
           "password": "123456",
           "database": "userinfo"
           }


class TaskConsumer(metaclass=ABCMeta):
    @abstractmethod
    def _task_worker(self, i):
        pass

    def consumer(self):
        task_pool = ThreadPool(self.task_pool_num)
        start_time = time.time()
        for i in range(self.task_num):
            task_pool.apply_async(func=self._task_worker, args=(i,))
        task_pool.close()
        task_pool.join()
        return time.time() - start_time

    def db_read(self, cur, sql, i):
        # time.sleep(2)
        cur.execute(sql)
        print(i, cur.fetchall())

class DbPoolTest(TaskConsumer):
    def __init__(self, task_num, task_pool_num, max_conn, blocking=True, **db_conf):
        self.task_num = task_num
        self.task_pool_num = task_pool_num
        self.max_conn=max_conn
        self.blocking = blocking
        self.kwargs = db_conf
        self._get_db_pool()

    def _get_db_pool(self):
        self.db_pool = pooled_db.PooledDB(
            creator=pymysql,  # 连接对象或符合DB-API 2的数据库模块（使用链接数据库的模块）
            mincached=self.max_conn,  # 池中空闲连接的初始数量（默认值为0表示启动时未建立连接）
            maxcached=0,  # 池中最大空闲连接数（默认值0或None表示池大小不受限制）
            maxshared=0,
            # 允许的最大共享连接数（默认值为0或None表示所有连接都是专用的）；达到此最大数目后，如果已将连接请求为可共享的，则将共享它们。因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
            maxconnections=self.max_conn,  # 通常允许的最大连接数（默认值为0或无表示任何数量的连接）
            blocking=self.blocking,  # 确定超出最大值时的行为（如果将其设置为true，阻塞并等待直到连接减少，但默认情况下将报告错误）
            maxusage=None,  # 单个连接的最大重用次数（默认值为0或无表示无限重用）；当达到连接的最大使用次数时，该连接将自动重置（关闭并重新打开）
            setsession=None,  # 可用于准备会话的SQL命令的可选列表，例如 [“将日期样式设置为德语”，...]
            reset=True,  # 将连接返回池时应如何重置（False或None为以begin（）开始的回滚事务，出于安全考虑，默认值True始终触发回滚）
            failures=None,  # 如果默认（OperationalError，InternalError）不足，则应为其应用连接故障转移机制的可选异常类或异常类的元组
            ping=1,
            # 可选标志，用于控制何时使用ping（）方法检查连接（如果该方法可用）（0 = None = never, 1 = default = whenever it is requested,2 = when a cursor is created, 4 = when a query is executed, 7 = always）
            **self.kwargs)

    def _task_worker(self, i):
        conn = self.db_pool.connection()
        cur = conn.cursor()
        sql = "select * from username;"
        self.db_read(cur, sql, i)
        cur.close()
        conn.close()

    def close(self):
        self.db_pool.close()

class PymysqlTest(TaskConsumer):
    def __init__(self,task_num, task_pool_num, **db_conf):
        self.task_num = task_num
        self.task_pool_num = task_pool_num
        self.kwargs = db_conf

    def _task_worker(self, i):
        conn = pymysql.Connect(**self.kwargs)
        cur = conn.cursor()
        sql = "select * from username;"
        self.db_read(cur, sql, i)
        cur.close()
        conn.close()


class PymysqlPoolTest(TaskConsumer):
    def __init__(self, task_num, task_pool_num, max_conn, blocking=True, **db_conf):
        self.task_num = task_num
        self.task_pool_num = task_pool_num
        self.max_conn = max_conn
        self.blocking = blocking
        self.kwargs = db_conf
        self._get_pymysql_pool()

    def _get_pymysql_pool(self):
        self.queue_pool = Queue(maxsize=self.max_conn)
        for i in range(self.max_conn):
            self.queue_pool.put(pymysql.Connect(**self.kwargs))

    def _task_worker(self, i):
        conn = self.queue_pool.get()
        cur = conn.cursor()
        sql = "select * from username;"
        self.db_read(cur, sql, i)
        cur.close()
        self.queue_pool.put(conn)

    def close(self):
        while not self.queue_pool.empty():
            self.queue_pool.get().close()

if __name__ == '__main__':
    TASK_NUM = 10000
    TASK_POOL_NUM = 100


    # 方案1：数据库连接池方案（建议设置blocking=True,maxconnections最大连接数量）
    start_time_dbpool=time.time()
    DB_POOL_NUM = 100 #DB_POOL_NUM>=TASK_POOL_NUM任务执行时间基本保持不变
    db_pool_test = DbPoolTest(TASK_NUM, TASK_POOL_NUM, DB_POOL_NUM,**db_conf)
    # start_time_dbpool = time.time()
    time_dbpool_task = db_pool_test.consumer()
    db_pool_test.close()
    time_dbpool = (time.time()-start_time_dbpool, time_dbpool_task)


    # 方案2：每个线程池中手动建立一个连接
    start_time_pymysql = time.time()
    pymysql_test = PymysqlTest(TASK_NUM, TASK_POOL_NUM, **db_conf)
    time_pymysql_task = pymysql_test.consumer()
    time_pymysql = (time.time()-start_time_pymysql, time_pymysql_task)

    # 方案3:手动维护数据库连接池
    start_time_pymysqlpool = time.time()
    DB_POOL_NUM = 100  # DB_POOL_NUM>=TASK_POOL_NUM任务执行时间基本保持不变
    pymysql_pool_test = PymysqlPoolTest(TASK_NUM, TASK_POOL_NUM, DB_POOL_NUM, **db_conf)
    # start_time_dbpool = time.time()
    time_pymysqlpool_task = pymysql_pool_test.consumer()
    pymysql_pool_test.close()
    time_pymysqlpool = (time.time() - start_time_pymysqlpool, time_pymysqlpool_task)


    print("数据库连接池方案：", time_dbpool)
    print("线程内手动连接方案：", time_pymysql)
    print("手动维护数据库连接池方案：", time_pymysqlpool)

数据库连接三种方案

DBUtils实际上是一个包含两个子模块的Python包，一个用于连接DB-API 2模块，另一个用于连接典型的PyGreSQL模块。