python封装数据库操作,python 连接池

2022-08-24 23:32:18 分类：python 阅读()

　　python封装数据库操作,python 连接池

　　本文主要介绍Python包数据库连接池的详细说明。文章重点介绍了题目的相关内容，有一定的参考价值，有需要的朋友可以参考一下。

　　00-1010一、数据库封装1.1数据库基本配置1.2编写singleton模式注释1.3构建连接池1.4封装MYsql的代码。二。连接池测试场景1:执行同一个实例两次。二。依次创建两个实例，并分别执行SQL。三。启动两个线程，但是创建连接池实例时有一个时间间隔。四。启动两个线程。创建连接池实例时，线程

　　线程安全：当两个线程同时使用线程池时，会同时创建两个线程池。如果多个线程交错使用线程池，那么只会创建一个线程池，并且共享一个线程池。我使用单例模式的注释方法。感觉就是这个注释的单实例方法，解决了多线程问题，但是没有解决线程安全问题。我需要优化这个单例模式。

　　主要通过PooledDB模块实现。

　　

目录

　　

一、数据库封装

　　前言：

　　# -*-编码： UTF-8 -*-

　　导入pymysql

　　#数据库信息

　　数据库测试主机=127.0.0.1

　　数据库测试端口=3308

　　DB_TEST_DBNAME=bt

　　DB_TEST_USER=root

　　DB_TEST_PASSWORD=123456

　　#数据库连接代码

　　DB_CHARSET=utf8

　　# mincached :启动时打开的空闲连接数(默认值0表示开始时不创建连接)

　　DB _ MIN _ CACHED=5

　　# maxcached :连接池中允许的最大空闲连接数(默认值0表示非空闲连接池的大小)

　　DB_MAX_CACHED=0

　　# maxshared :允许的最大共享连接数(默认值0表示所有连接都是专用的)。如果达到最大数量，请求共享的连接将被共享。

　　DB_MAX_SHARED=5

　　# MaxConnections:创建的连接池的最大数量(默认值0表示无限制)

　　DB _ MAX _ CONNECYIONS=300

　　# blocking :设置连接池达到最大数量时的行为(默认值0或False表示返回一个错误给Many.其他意味着阻塞，直到连接数减少并且连接被分配)

　　DB_BLOCKING=True

　　# maxusage :单个连接允许的最大重用次数(默认值0或False表示无限制重用)。当达到最大数量时，连接将自动重新连接(关闭并重新打开)

　　DB_MAX_USAGE=0

　　# set session : SQL命令的可选列表用于准备每个会话，例如[set datestyle to german ，]

　　DB_SET_SESSION=无

　　# Creator3360使用连接到数据库的模块

　　DB_CREATOR=pymysql

　　将最大和最小连接池设置为5。当连接池启动时，将建立五个连接。

　　

1.1数据库基本配置

　　db_config.py

　　# singleton模式函数，用于装饰类

　　def singleton(cls，*args，**kw):

　　实例={}

　　def _singleton():

　　如果cls不在实例：中

　　实例[cls]=cls(*args，**kw)

　　返回实例[cls]

　　return _ singleton

　　

1.2 编写单例模式注解

　　singleton.py

　　从dbutils.pooled_db导入PooledDB

　　将db_config作为配置导入

　　#

　　import random

　　from singleton import singleton

　　"""

　　@功能：创建数据库连接池

　　"""

　　class MyConnectionPool(object):

　　 # 私有属性

　　 # 能通过对象直接访问，但是可以在本类内部访问；

　　 __pool = None

　　 # def __init__(self):

　　 # self.conn = self.__getConn()

　　 # self.cursor = self.conn.cursor()

　　 # 创建数据库连接conn和游标cursor

　　 def __enter__(self):

　　 self.conn = self.__getconn()

　　 self.cursor = self.conn.cursor()

　　 # 创建数据库连接池

　　 def __getconn(self):

　　 if self.__pool is None:

　　 # i = random.randint(1, 100)

　　 # print("创建线程池的数量"+str(i))

　　 self.__pool = PooledDB(

　　 creator=config.DB_CREATOR,

　　 mincached=config.DB_MIN_CACHED,

　　 maxcached=config.DB_MAX_CACHED,

　　 maxshared=config.DB_MAX_SHARED,

　　 maxconnections=config.DB_MAX_CONNECYIONS,

　　 blocking=config.DB_BLOCKING,

　　 maxusage=config.DB_MAX_USAGE,

　　 setsession=config.DB_SET_SESSION,

　　 host=config.DB_TEST_HOST,

　　 port=config.DB_TEST_PORT,

　　 user=config.DB_TEST_USER,

　　 passwd=config.DB_TEST_PASSWORD,

　　 db=config.DB_TEST_DBNAME,

　　 use_unicode=False,

　　 charset=config.DB_CHARSET

　　 )

　　 return self.__pool.connection()

　　 # 释放连接池资源

　　 def __exit__(self, exc_type, exc_val, exc_tb):

　　 self.cursor.close()

　　 self.conn.close()

　　 # 关闭连接归还给链接池

　　 # def close(self):

　　 # self.cursor.close()

　　 # self.conn.close()

　　 # 从连接池中取出一个连接

　　 def getconn(self):

　　 conn = self.__getconn()

　　 cursor = conn.cursor()

　　 return cursor, conn

　　# 获取连接池,实例化

　　@singleton

　　def get_my_connection():

　　 return MyConnectionPool()

　　

　　

1.4封装Python操作MYSQL的代码

　　mysqlhelper.py

　　

import time
　　from db_dbutils_init import get_my_connection
　　"""执行语句查询有结果返回结果没有返回0；增/删/改返回变更数据条数，没有返回0"""
　　class MySqLHelper(object):
　　 def __init__(self):
　　 self.db = get_my_connection() # 从数据池中获取连接
　　 #
　　 # def __new__(cls, *args, **kwargs):
　　 # if not hasattr(cls, inst): # 单例
　　 # cls.inst = super(MySqLHelper, cls).__new__(cls, *args, **kwargs)
　　 # return cls.inst
　　 # 封装执行命令
　　 def execute(self, sql, param=None, autoclose=False):
　　 """
　　 【主要判断是否有参数和是否执行完就释放连接】
　　 :param sql: 字符串类型，sql语句
　　 :param param: sql语句中要替换的参数"select %s from tab where id=%s" 其中的%s就是参数
　　 :param autoclose: 是否关闭连接
　　 :return: 返回连接conn和游标cursor
　　 """
　　 cursor, conn = self.db.getconn() # 从连接池获取连接
　　 count = 0
　　 try:
　　 # count : 为改变的数据条数
　　 if param:
　　 count = cursor.execute(sql, param)
　　 else:
　　 count = cursor.execute(sql)
　　 conn.commit()
　　 if autoclose:
　　 self.close(cursor, conn)
　　 except Exception as e:
　　 pass
　　 return cursor, conn, count
　　 # 释放连接
　　 def close(self, cursor, conn):
　　 """释放连接归还给连接池"""
　　 cursor.close()
　　 conn.close()
　　 # 查询所有
　　 def selectall(self, sql, param=None):
　　 cursor = None
　　 conn = None
　　 count = None
　　 try:
　　 cursor, conn, count = self.execute(sql, param)
　　 res = cursor.fetchall()
　　 return res
　　 except Exception as e:
　　 print(e)
　　 self.close(cursor, conn)
　　 return count
　　 # 查询单条
　　 def selectone(self, sql, param=None):
　　 cursor = None
　　 conn = None
　　 count = None
　　 try:
　　 cursor, conn, count = self.execute(sql, param)
　　 res = cursor.fetchone()
　　 self.close(cursor, conn)
　　 return res
　　 except Exception as e:
　　 print("error_msg:", e.args)
　　 self.close(cursor, conn)
　　 return count
　　 # 增加
　　 def insertone(self, sql, param):
　　 cursor = None
　　 conn = None
　　 count = None
　　 try:
　　 cursor, conn, count = self.execute(sql, param)
　　 # _id = cursor.lastrowid() # 获取当前插入数据的主键id，该id应该为自动生成为好
　　 conn.commit()
　　 self.close(cursor, conn)
　　 return count
　　 except Exception as e:
　　 print(e)
　　 conn.rollback()
　　 self.close(cursor, conn)
　　 return count
　　 # 增加多行
　　 def insertmany(self, sql, param):
　　 """
　　 :param sql:
　　 :param param: 必须是元组或列表[(),()]或（（），（））
　　 :return:
　　 """
　　 cursor, conn, count = self.db.getconn()
　　 try:
　　 cursor.executemany(sql, param)
　　 conn.commit()
　　 return count
　　 except Exception as e:
　　 print(e)
　　 conn.rollback()
　　 self.close(cursor, conn)
　　 return count
　　 # 删除
　　 def delete(self, sql, param=None):
　　 cursor = None
　　 conn = None
　　 count = None
　　 try:
　　 cursor, conn, count = self.execute(sql, param)
　　 self.close(cursor, conn)
　　 return count
　　 except Exception as e:
　　 print(e)
　　 conn.rollback()
　　 self.close(cursor, conn)
　　 return count
　　 # 更新
　　 def update(self, sql, param=None):
　　 cursor = None
　　 conn = None
　　 count = None
　　 try:
　　 cursor, conn, count = self.execute(sql, param)
　　 conn.commit()
　　 self.close(cursor, conn)
　　 return count
　　 except Exception as e:
　　 print(e)
　　 conn.rollback()
　　 self.close(cursor, conn)
　　 return count
　　# if __name__ == __main__:
　　# db = MySqLHelper()
　　# sql = "SELECT SLEEP(10)"
　　# db.execute(sql)
　　# time.sleep(20)
　　 # TODO 查询单条
　　 # sql1 = select * from userinfo where name=%s
　　 # args = python
　　 # ret = db.selectone(sql=sql1, param=args)
　　 # print(ret) # (None, bpython, b123456, b0)
　　 # TODO 增加单条
　　 # sql2 = insert into hotel_urls(cname,hname,cid,hid,url) values(%s,%s,%s,%s,%s)
　　 # ret = db.insertone(sql2, (1, 2, 1, 2, 2))
　　 # print(ret)
　　 # TODO 增加多条
　　 # sql3 = insert into userinfo (name,password) VALUES (%s,%s)
　　 # li = li = [
　　 # (分省, 123),
　　 # (到达,456)
　　 # ]
　　 # ret = db.insertmany(sql3,li)
　　 # print(ret)
　　 # TODO 删除
　　 # sql4 = delete from userinfo WHERE name=%s
　　 # args = xxxx
　　 # ret = db.delete(sql4, args)
　　 # print(ret)
　　 # TODO 更新
　　 # sql5 = rupdate userinfo set password=%s WHERE name LIKE %s
　　 # args = (993333993, %old%)
　　 # ret = db.update(sql5, args)
　　 # print(ret)

　　

　　

二、连接池测试

　　修改db_dbutils_init.py 文件，在创建连接池def __getconn(self):方法下，加一个打印随机数，方便将来我们定位是否时单例的线程池。

　　

　　修改后的db_dbutils_init.py 文件：

　　

from dbutils.pooled_db import PooledDB
　　import db_config as config
　　import random
　　from singleton import singleton
　　"""
　　@功能：创建数据库连接池
　　"""
　　class MyConnectionPool(object):
　　 # 私有属性
　　 # 能通过对象直接访问，但是可以在本类内部访问；
　　 __pool = None
　　 # def __init__(self):
　　 # self.conn = self.__getConn()
　　 # self.cursor = self.conn.cursor()
　　 # 创建数据库连接conn和游标cursor
　　 def __enter__(self):
　　 self.conn = self.__getconn()
　　 self.cursor = self.conn.cursor()
　　 # 创建数据库连接池
　　 def __getconn(self):
　　 if self.__pool is None:
　　 i = random.randint(1, 100)
　　 print("线程池的随机数"+str(i))
　　 self.__pool = PooledDB(
　　 creator=config.DB_CREATOR,
　　 mincached=config.DB_MIN_CACHED,
　　 maxcached=config.DB_MAX_CACHED,
　　 maxshared=config.DB_MAX_SHARED,
　　 maxconnections=config.DB_MAX_CONNECYIONS,
　　 blocking=config.DB_BLOCKING,
　　 maxusage=config.DB_MAX_USAGE,
　　 setsession=config.DB_SET_SESSION,
　　 host=config.DB_TEST_HOST,
　　 port=config.DB_TEST_PORT,
　　 user=config.DB_TEST_USER,
　　 passwd=config.DB_TEST_PASSWORD,
　　 db=config.DB_TEST_DBNAME,
　　 use_unicode=False,
　　 charset=config.DB_CHARSET
　　 )
　　 return self.__pool.connection()
　　 # 释放连接池资源
　　 def __exit__(self, exc_type, exc_val, exc_tb):
　　 self.cursor.close()
　　 self.conn.close()
　　 # 关闭连接归还给链接池
　　 # def close(self):
　　 # self.cursor.close()
　　 # self.conn.close()
　　 # 从连接池中取出一个连接
　　 def getconn(self):
　　 conn = self.__getconn()
　　 cursor = conn.cursor()
　　 return cursor, conn
　　 # 获取连接池,实例化
　　@singleton
　　def get_my_connection():
　　 return MyConnectionPool()

　　开始测试：

　　

　　

场景一：同一个实例，执行2次sql

　　

from mysqlhelper import MySqLHelper
　　import time
　　if __name__ == __main__:
　　 sql = "SELECT SLEEP(10)"
　　 sql1 = "SELECT SLEEP(15)"
　　 db = MySqLHelper()
　　 db.execute(sql)
　　 db.execute(sql1)
　　 time.sleep(20)

　　在数据库中，使用show processlist;

　　

show processlist;

　　当执行第一个sql时。数据库连接显示。

　　

　　当执行第二个sql时。数据库连接显示：

　　

　　当执行完sql，程序sleep时。数据库连接显示：

　　

　　程序打印结果：

　　

线程池的随机数43

　　

　　由以上可以得出结论：

　　线程池启动后，生成了5个连接。执行第一个sql时，使用了1个连接。执行完第一个sql后，使用了另外1个连接。 这是一个线性的，线程池中一共5个连接，但是每次执行，只使用了其中一个。

　　有个疑问，连接池如果不支持并发是不是就毫无意义？

　　如上，虽然开了线程池5个连接，但是每次执行sql，只用到了一个连接。那为何不设置线程池大小为1呢？设置线程池大小的意义何在呢？（如果在非并发的场景下，是不是设置大小无意义？）

　　相比于不用线程池的优点：

　　如果不用线程池，则每次执行一个sql都要创建、断开连接。 像我们这样使用连接池，不用反复创建、断开连接，拿现成的连接直接用就好了。

　　

　　

场景二：依次创建2个实例，各自执行sql

　　

from mysqlhelper import MySqLHelper
　　import time
　　if __name__ == __main__:
　　 db = MySqLHelper()
　　 db1 = MySqLHelper()
　　 sql = "SELECT SLEEP(10)"
　　 sql1 = "SELECT SLEEP(15)"
　　 db.execute(sql)
　　 db1.execute(sql1)
　　 time.sleep(20)

　　第一个实例db，执行sql。线程池启动了5个连接

　　

　　第二个实例db1，执行sql：

　　

　　程序睡眠时，一共5个线程池：

　　

　　打印结果：

　　

　　结果证明：

　　虽然我们依次创建了2个实例，但是（1）创建线程池的打印结果，只打印1次，且从始至终，线程池一共只启动了5个连接，且连接的id没有发生改变，说明一直是这5个连接。

　　证明，我们虽然创建了2个实例，但是这2个实例其实是一个实例。（单例模式是生效的）

　　

　　

场景三：启动2个线程，但是线程在创建连接池实例时，有时间间隔

　　

import threading
　　from mysqlhelper import MySqLHelper
　　import time
　　def sl1():
　　 time.sleep(2)
　　 db = MySqLHelper()
　　 sql = "SELECT SLEEP(6)"
　　 db.execute(sql)
　　def sl2():
　　 time.sleep(4)
　　 db = MySqLHelper()
　　 sql = "SELECT SLEEP(15)"
　　 db.execute(sql)
　　if __name__ == __main__:
　　 threads = []
　　 t1 = threading.Thread(target=sl1)
　　 threads.append(t1)
　　 t2 = threading.Thread(target=sl2)
　　 threads.append(t2)
　　 for t in threads:
　　 t.setDaemon(True)
　　 t.start()
　　 time.sleep(20)

　　2个线程间隔了2秒。

　　观察数据库的连接数量：

　　

　　打印结果：

　　

　　在并发执行2个sql时，共用了这5个连接，且打印结果只打印了一次，说明虽然并发创建了2次实例，但真正只创建了一个连接池。

　　

　　

场景四：启动2个线程，线程在创建连接池实例时，没有时间间隔

　　

import threading
　　from mysqlhelper import MySqLHelper
　　import time
　　if __name__ == __main__:
　　 db = MySqLHelper()
　　 sql = "SELECT SLEEP(6)"
　　 sql1 = "SELECT SLEEP(15)"
　　 threads = []
　　 t1 = threading.Thread(target=db.execute, args=(sql,))
　　 threads.append(t1)
　　 t2 = threading.Thread(target=db.execute, args=(sql1,))
　　 threads.append(t2)
　　 for t in threads:
　　 t.setDaemon(True)
　　 t.start()
　　 time.sleep(20)

　　观察数据库连接 ：

　　

　　打印结果：

　　

　　结果表明：

　　终端打印了2次，数据库建立了10个连接，说明创建了2个线程池。这样的单例模式，存在线程安全问题。

　　到此这篇关于Python封装数据库连接池详解的文章就介绍到这了,更多相关Python连接池内容请搜索盛行IT软件开发工作室以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT软件开发工作室！

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。

标签： python

相关文章阅读

  • 关系型数据库与非关系型数据库简介一样吗,关系型数据库非关系型数据库有哪些
  • 关系型数据库与非关系型数据库简介一样吗,关系型数据库非关系型数据库有哪些,关系型数据库与非关系型数据库简介
  • 关于redis数据库入门详细介绍图片,redis数据库的使用,关于Redis数据库入门详细介绍
  • 使用php连接mysql数据库,php连接数据库的方法
  • 使用php连接mysql数据库,php连接数据库的方法,一文详解PHP连接MySQL数据库的三种方式
  • 什么是分库分表,为什么要进行分库分表-,分库分表的区别,数据库分库分表是什么，什么情况下需要用分库分表
  • vb中adodb连接数据库,
  • treeview控件绑定数据,wpf treeview数据绑定,详解TreeView绑定数据库
  • sql的多表查询,数据库如何实现多表查询
  • SQL数据库的图形管理界面工具是,sql图形界面创建数据库
  • SQL数据库的图形管理界面工具是,sql图形界面创建数据库,SQLServer2019 数据库的基本使用之图形化界面操作的实现
  • sql数据库定时备份怎么弄,mysql 定期备份
  • sql数据库定时备份怎么弄,mysql 定期备份,MySQL 数据库定时备份的几种方式(全面)
  • sqlserver的nvarchar和varchar,数据库varchar和nvarchar
  • sqlserver的nvarchar和varchar,数据库varchar和nvarchar,SQL中varchar和nvarchar的基本介绍及其区别
    • 留言与评论（共有 条评论）
       
    验证码：