数据库连接池

# @author： zhc
# @Time: 2023/5/5
# @FileName: 数据库连接池
# pip install PyMySQL , pip install DBUtils==1.3
import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模板
    maxconnections=6,
    mincached=2,
    # maxcached=5,
    # maxshared=3,
    blocking=True,
    # maxusage=None,
    # setsession=[],
    # ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='p@ss0rd',
    # database='pooldb',
    database='cov',
    charset='utf8'
)

# 去连接池中获取一个连接
conn = POOL.connection()

cursor = conn.cursor()
cursor.execute('select * from drug')
result = cursor.fetchall()

cursor.close()

# 讲连接放回到连接池
conn.close()

print(result)

多线程测试

# @author： zhc
# @Time: 2023/5/5
# @FileName: 数据库连接池
# pip3 install PyMySQL , pip install DBUtils==1.3
import pymysql
from threading import Thread
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模板
    maxconnections=6,
    mincached=2,
    # maxcached=5,
    # maxshared=3,
    blocking=True,
    # maxusage=None,
    # setsession=[],
    # ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='p@ss0rd',
    # database='pooldb',
    database='cov',
    charset='utf8'
)




def task(num):
    # 去连接池中获取一个连接
    conn = POOL.connection()

    cursor = conn.cursor()
    # cursor.execute('select * from drug')
    cursor.execute('select sleep(3)')
    result = cursor.fetchall()

    cursor.close()

    # 讲连接放回到连接池
    conn.close()
    print(num, "-------------->", result)



for i in range(57):
    t = Thread(target=task, args=(i,))
    t.start()

函数实现sqlhelper

# @author： zhc
# @Time: 2023/5/5
# @FileName: sqlhelper

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模板
    maxconnections=6,
    mincached=2,
    # maxcached=5,
    # maxshared=3,
    blocking=True,
    # maxusage=None,
    # setsession=[],
    # ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='p@ss0rd',
    # database='pooldb',
    database='cov',
    charset='utf8'
)

def fetchall(sql, *args):
    """获取所有数据"""
    # 去连接池中获取一个连接
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute(sql, args)
    result = cursor.fetchall()
    cursor.close()
    conn.close()

    return result

def fetchone(sql, *args):
    """获取单条数据"""
    # 去连接池中获取一个连接
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute(sql, args)
    result = cursor.fetchone()
    cursor.close()
    conn.close()

    return result

单例模式

# @author： zhc
# @Time: 2023/5/5
# @FileName: sqlhelper2

import pymysql
from DBUtils.PooledDB import PooledDB

class SqlHelper:
    def __init__(self):
        self.pool = PooledDB(
                    creator=pymysql,  # 使用链接数据库的模板
                    maxconnections=6,
                    mincached=2,
                    blocking=True,
                    host='127.0.0.1',
                    port=3306,
                    user='root',
                    password='p@ss0rd',
                    # database='pooldb',
                    database='cov',
                    charset='utf8'
                )


    def __open(self):
        conn = self.pool.connection()
        cursor = conn.cursor()
        return conn, cursor

    def __close(self, cursor, conn):
        cursor.close()
        conn.close()


    def fetchall(self, sql, *args):
        """获取所有数据"""
        # 去连接池中获取一个连接
        conn, cursor = self.__open()
        cursor.execute(sql, args)
        result = cursor.fetchall()
        self.__close(cursor, conn)

        return result

    def fetchone(self, sql, *args):
        """获取单条数据"""
        # 去连接池中获取一个连接
        conn, cursor = self.__open()
        cursor.execute(sql, args)
        result = cursor.fetchone()
        self.__close(cursor, conn)

        return result


db = SqlHelper()