零、综述：

python 封装myslq、oracle连接池。

但是现在有个bug，就是线程不安全的问题，如果多个线程，同时启动，会创建多个线程池。（如果多个线程，不同时启动，会共用一个线程池），后续会优化这个问题。

这里不是简单粗暴的封装，而是加了一点点的设计模式。不是专业的，见谅。

一、python封装myslq连接池

1、数据库连接信息，单独存放在一个yml文件

2、自定义类实体类DbInfo用来存储数据库连接信息

3、自定义方法readDbYml，读取yml文件，转成DbInfo

4、连接池配置信息也单独存放在了一个py文件。

1.1数据库连接信息yml文件

a.yml

DB:host: 1.1.1.1username: namepassword: pwdatabase: databasenamedatabasetype: mysqlport: 3306note: 备注

1.2自定义类实体类DbInfo用来存储数据库连接信息

DbInfo.py

class DbInfo():# 在构造器里直接赋值，好像就不要写属性了。省了get和set方法了？def __init__(self,host=None,username=None,password=None,database=None,databasetype=None,port=None,note=None,charset="utf8"):self.host: str = hostself.username: str= usernameself.password: str = passwordself.database: str = databaseself.databasetype: str = databasetypeself.port: str = portself.note:str = noteself.charset:str = charset

1.3 自定义方法readDbYml，读取连接信息yml文件，转成DbInfo

YmlUtil.py


import yaml
from model.DbInfo import *
class YmlUtil():# path = os.path.abspath()@staticmethoddef readDbYml(filePath:str) -> DbInfo:f = open(filePath, 'r', encoding='utf-8')cont = f.read()x = yaml.load(cont,Loader=yaml.FullLoader)print(x['DB'])print(x['DB']['host'])dbInfo=DbInfo(host=x['DB']['host'],username=x['DB']['username'],password=x['DB']['password'],database=x['DB']['database'],port=x['DB']['port'])# 设置类属性——setattr(object,key,value)# 类似于建造者模式setattr(dbInfo,"note",x['DB']['note'])return dbInfo

1.4 连接池配置

DBPoolSet.py

import pymysql
class DBPoolSet:"""数据库连接池设置"""# 属性# 数据库连接编码DB_CHARSET = "utf8"# mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)DB_MIN_CACHED = 3# maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小)DB_MAX_CACHED = 3# maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用DB_MAX_SHARED = 2# maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)DB_MAX_CONNECYIONS = 3# blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......> 其他代表阻塞直到连接数减少,连接被分配)DB_BLOCKING = True# maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)DB_MAX_USAGE = 0# setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...]DB_SET_SESSION = None# creator : 使用连接数据库的模块DB_CREATOR = pymysql

1.5 单例模式注解

singleton.py

def singleton(cls,*args,**kw):instances = {}def _singleton():if cls not in instances:instances[cls] = cls(*args,**kw)return instances[cls]return _singleton

1.6 封装数据库实体类ConnectionPool.py

ConnectionPool.py

from dbutils.pooled_db import PooledDB
from DBPoolSet import DBPoolSet
from model.DbInfo import DbInfo
import random
import osfrom utils.YmlUtil import YmlUtil
from utils.singleton import singleton"""
@功能：创建数据库连接池
"""class ConnectionPool(object):__pool = Nonepath = "/Users/zhaohui/a.yml"dbInfo = YmlUtil.readDbYml(path)dbPoolSet = DBPoolSet()# 构造器，传DbInfo、DBPoolSet# def __init__(self):#     self.conn = self.__getconn()#     self.cursor = self.conn.cursor()# 创建数据库连接conn和游标cursordef __enter__(self):self.conn = self.__getconn()self.cursor = self.conn.cursor()# 创建数据库连接池,私有方法def __getconn(self):if self.__pool is None:self.__pool = PooledDB(creator=self.dbPoolSet.DB_CREATOR, # creator : 使用连接数据库的模块mincached=self.dbPoolSet.DB_MIN_CACHED, # mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)maxcached=self.dbPoolSet.DB_MAX_CACHED,maxshared=self.dbPoolSet.DB_MAX_SHARED,maxconnections=self.dbPoolSet.DB_MAX_CONNECYIONS,blocking=self.dbPoolSet.DB_BLOCKING,maxusage=self.dbPoolSet.DB_MAX_USAGE,setsession=self.dbPoolSet.DB_SET_SESSION,host=self.dbInfo.host,port=self.dbInfo.port,user=self.dbInfo.username,passwd=self.dbInfo.password,db=self.dbInfo.database,use_unicode=False,charset=self.dbInfo.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_connection():return ConnectionPool()

1.7 封装数据库操作

HandleSql.py

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 06 -09"""
执行语句查询有结果返回结果没有返回0；
增/删/改返回变更数据条数，没有返回0
"""class HandleSql:__instance = Nonedef __init__(self):self.db = get_connection()# 封装执行命令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 = 0try:# 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:passreturn cursor, conn, count# 释放连接def close(self, cursor, conn):"""释放连接归还给连接池"""cursor.close()conn.close()# 查询所有def selectall(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)res = cursor.fetchall()return resexcept Exception as e:print(e)self.close(cursor, conn)return count# 查询单条def selectone(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)res = cursor.fetchone()self.close(cursor, conn)return resexcept Exception as e:print("error_msg:", e.args)self.close(cursor, conn)return count# 增加def insertone(self, sql, param):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)# _id = cursor.lastrowid()  # 获取当前插入数据的主键id，该id应该为自动生成为好conn.commit()self.close(cursor, conn)return countexcept 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 countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# 删除def delete(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)self.close(cursor, conn)return countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# 更新def update(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)conn.commit()self.close(cursor, conn)return countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# TODO 查询单条
# sql1 = 'select * from userinfo where name=%s'
# args = 'python'
# ret = db.selectone(sql=sql1, param=args)
# print(ret)  # (None, b'python', b'123456', b'0')# 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 = r'update userinfo set password=%s WHERE name LIKE %s'
# args = ('993333993', '%old%')
# ret = db.update(sql5, args)
# print(ret)# 调用方式
# if __name__ == '__main__':
#     a = KyfActivityHandleSql()
#     list=['15555555555']
#     x=a.delete_activity(list)
#     print(x)
#

二、python封装oracle连接池

1、数据库连接信息，单独存放在一个yml文件

2、自定义类实体类DbInfo用来存储数据库连接信息

3、自定义方法readDbYml，读取yml文件，转成DbInfo

4、连接池配置信息也单独存放在了一个py文件。

2.1 数据库连接信息yml文件

b.yml

DB:host: 2.2.2.2username: namepassword: pwdatabase: servicenamedatabasetype: ORCLDGport: 3306note: 备注

注意：database: servicename ，database 这里填写的不是库的名字，而是servicename

2.2自定义类实体类DbInfo用来存储数据库连接信息

2.2 与1.2是重复的，写一次就行

DbInfo.py

class DbInfo():# 在构造器里直接赋值，好像就不要写属性了。省了get和set方法了？def __init__(self,host=None,username=None,password=None,database=None,databasetype=None,port=None,note=None,charset="utf8"):self.host: str = hostself.username: str= usernameself.password: str = passwordself.database: str = databaseself.databasetype: str = databasetypeself.port: str = portself.note:str = noteself.charset:str = charset

2.3 自定义方法readDbYml，读取连接信息yml文件，转成DbInfo

2.3 与1.3是重复的，写一次就行

YmlUtil.py


import yaml
from model.DbInfo import *
class YmlUtil():# path = os.path.abspath()@staticmethoddef readDbYml(filePath:str) -> DbInfo:f = open(filePath, 'r', encoding='utf-8')cont = f.read()x = yaml.load(cont,Loader=yaml.FullLoader)print(x['DB'])print(x['DB']['host'])dbInfo=DbInfo(host=x['DB']['host'],username=x['DB']['username'],password=x['DB']['password'],database=x['DB']['database'],port=x['DB']['port'])# 设置类属性——setattr(object,key,value)# 类似于建造者模式setattr(dbInfo,"note",x['DB']['note'])return dbInfo

2.4 连接池配置

OraclePoolSet.py

class OraclePoolSet:"""数据库连接池设置"""# 属性# 连接池大小pool_size = 3

2.5 单例模式注解

2.5 与1.5是重复的，写一次就行

singleton.py

def singleton(cls,*args,**kw):instances = {}def _singleton():if cls not in instances:instances[cls] = cls(*args,**kw)return instances[cls]return _singleton

2.6 封装数据库实体类OracleConnectionPool.py

OracleConnectionPool.py

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 06 -28import osimport cx_Oracle as Oraclefrom OraclePoolSet import OraclePoolSet
from utils.YmlUtil import YmlUtil
from utils.singleton import singletonclass OracleConnectionPool(object):__pool = Nonepath = "/Users/conf/dbinfo/b.yml"dbInfo = YmlUtil.readDbYml(path)dbPoolSet = OraclePoolSet()# 创建数据库连接conn和游标cursordef __enter__(self):self.conn = self.__getconn()self.cursor = self.conn.cursor()# 创建数据库连接池,私有方法def __getconn(self):if self.__pool is None:self.__pool = Oracle.SessionPool(user=str(self.dbInfo.username),password=str(self.dbInfo.password),dsn=Oracle.makedsn(str(self.dbInfo.host),str(self.dbInfo.port),service_name=str(self.dbInfo.database)),min=self.dbPoolSet.pool_size,max=self.dbPoolSet.pool_size,increment=0,encoding="UTF-8")# return self.__pool.connection()return self.__pool# 释放连接池资源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().acquire()cursor = conn.cursor()return cursor, conn
# 获取连接池,实例化
@singleton
def get_oracle_connection():return OracleConnectionPool()

2.7 封装数据库操作

OracleHandleSql.py

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 06 -28import reclass KDHandleSql:__instance = Nonedef __init__(self):self.db = get_oracle_connection()# 封装执行命令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 = 0try:# 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:passreturn cursor, conn, count# 释放连接def close(self, cursor, conn):"""释放连接归还给连接池"""cursor.close()conn.close()# 查询所有def selectall(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)res = cursor.fetchall()return resexcept Exception as e:print(e)self.close(cursor, conn)return count# 查询单条def selectone(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)res = cursor.fetchone()self.close(cursor, conn)return resexcept Exception as e:print("error_msg:", e.args)self.close(cursor, conn)return count# 增加def insertone(self, sql, param):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)# _id = cursor.lastrowid()  # 获取当前插入数据的主键id，该id应该为自动生成为好conn.commit()self.close(cursor, conn)return countexcept 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 countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# 删除def delete(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)self.close(cursor, conn)return countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# 更新def update(self, sql, param=None):cursor = Noneconn = Nonecount = Nonetry:cursor, conn, count = self.execute(sql, param)conn.commit()self.close(cursor, conn)return countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count

三、参考连接：

python3 mysql 连接池_yFwillh的博客-CSDN博客_python3 数据库连接池

原文存在多线程问题，多线程会创建多个连接池。

我修改后，存在线程安全问题。当2个线程同时用到线程池时，会同时创建2个线程池。如果多个线程，错开用到线程池，就只会创建一个线程池，会共用一个线程池。

我用的注解方式的单例模式，感觉就是这个注解的单例方式，解决了多线程问题，但是没解决线程安全问题，需要优化这个单例模式。

四、测试一下连接池是否成功（可以忽略这一段）

4.1、连接池测试

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

修改后的db_dbutils_init.py 文件

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 06 -19
# @File: db_dbutils_init.py
from dbutils.pooled_db import PooledDB
import db_config as config
import randomfrom singleton import singleton"""
@功能：创建数据库连接池
"""class MyConnectionPool(object):# 私有属性# 能通过对象直接访问，但是可以在本类内部访问；__pool = None# def __init__(self):#     self.conn = self.__getConn()#     self.cursor = self.conn.cursor()# 创建数据库连接conn和游标cursordef __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 timeif __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 timeif __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个线程，但是线程在创建连接池实例时，有时间间隔

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 06 -19
# @File: test1.py
# 并发执行
import threadingfrom mysqlhelper import MySqLHelper
import timedef 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个线程，线程在创建连接池实例时，没有时间间隔

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 06 -19
# @File: testconnect.py
import threadingfrom mysqlhelper import MySqLHelper
import timeif __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个线程池。

这样的单例模式，存在线程安全问题。

4.3、问题&解决问题

如上面的实验，存在线程不安全问题。根本原因，应该就出在了单例模式那儿。

等我想想要怎么应用这个单例模式。

Python封装mysqloracle数据库连接池相关推荐

Python操作数据库及Python实现mysql数据库连接池源代码
简介 pymysql:纯Python实现的一个驱动.因为是纯Python编写的,因此执行效率不如MySQL-python.并且也因为是纯Python编写的,因此可以和Python代码无缝衔接. MyS ...
【Python SQLAlchemy】数据库连接池SQLAlchemy中多线程安全的问题
问题描述写了一个apscheduler定时任务里面用到SQLAlchemy在同一时间点开启了18个任务,用于更新18个表但是最终看数据库表的时候,发现有2个表未更新.查看打印日志均正常,查找原因 ...
Python 多进程与数据库连接池配合同时取出数据进行处理
1 引入问题:曾经想过用分布式来处理,但是似乎并不需要那么麻烦,使用Python 数据库连接池与Python 进程池,同时对数据的读取与处理. 这里就不赘述连接池怎么建立了,主要是Python 进程池 ...
Python下mysql数据库连接池
文章目录前言使用模块创建连接池取出连接连接放回池子前言参考我的文章 Python操作mysql 连接池能复用mysql连接,节约资源使用模块 import mysql.connecto ...
在使用 python 封装的进程池 from concurrent.futures import ProcessPoolExecutor 遇到的问题
在ubuntu中,用的是python3.5 executeprebuildpath = ExecutePrebuild()processpool = ProcessPoolExecutor(1)pro ...
python实现数据库连接池_Python实现Mysql数据库连接池
Python实现Mysql数据库连接池 python连接Mysql数据库: python编程中可以使用MySQLdb进行数据库的连接及诸如查询/插入/更新等操作,但是每次连接mysql数据库请求时,都 ...
Python 数据库连接方法和数据库连接池
Python连接数据库 Python官方制定的数据库接口标准中,主要包含了顶层connect函数.部分常量.数据库操作异常.用于管理连接的Connection类以及执行查询的Cursor类. DBUt ...
c3p0，DBCP，Druid（德鲁伊）数据库连接池
c3p0,DBCP,Druid(德鲁伊)数据库连接池每博一文案佛说:前世 500 次的回眸,才换来今生的一次擦肩而过. 人与人之间的缘分,真的无需强求,并不是所有的感情都能天长地久,正如<越 ...
python数据库连接池工具类_Python数据库连接池DBUtils
DBUtils简介 DBUtils是Python的一个用于实现数据库连接池的模块. 此连接池有两种连接模式: 模式一:为每个线程创建一个连接,线程即使调用了close方法,也不会关闭,只是把连接重新放 ...

Python封装mysqloracle数据库连接池

零、综述：

一、python封装myslq连接池

1.1数据库连接信息yml文件

1.2自定义类实体类DbInfo用来存储数据库连接信息

1.3 自定义方法readDbYml，读取连接信息yml文件，转成DbInfo

1.4 连接池配置

1.5 单例模式注解

1.6 封装数据库实体类ConnectionPool.py

1.7 封装数据库操作

二、python封装oracle连接池

2.1 数据库连接信息yml文件

2.2自定义类实体类DbInfo用来存储数据库连接信息

2.3 自定义方法readDbYml，读取连接信息yml文件，转成DbInfo

2.4 连接池配置

2.5 单例模式注解

2.6 封装数据库实体类OracleConnectionPool.py

2.7 封装数据库操作

三、参考连接：

四、测试一下连接池是否成功（可以忽略这一段）

4.1、连接池测试

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

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

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

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

4.3、问题&解决问题

Python封装mysqloracle数据库连接池相关推荐

最新文章

热门文章