1.主要模块
class DbPool(object)：数据库连接池类
class ThreadDb(DbPool)：多线程操作类
2.数据库连接池PooledDB
PooledDB 基本参数：
mincached : 最少的空闲连接数，如果空闲连接数小于这个数，Pool自动创建新连接;
maxcached : 最大的空闲连接数，如果空闲连接数大于这个数，Pool则关闭空闲连接;
maxconnections : 最大的连接数;
blocking : 当连接数达到最大的连接数时，在请求连接的时候，如果这个值是True，请求连接的程序会一直等待，直到当前连接数小于最大连接数，如果这个值是False，会报错;
更多的参数介绍请参考：https://blog.csdn.net/diuleilaomu/article/details/103278147
3，MySQL表结构如下图:
4，全部代码：
Code DbPool:
class DbPool(object):
def __init__(self,db_type,config):
self.__db_type = db_type
if self.__db_type =='mysql':
db_creator = importlib.import_module('pymysql')
elif self.__db_type =='sqlserver':
db_creator = importlib.import_module('pymssql')
elif self.__db_type =='oracle':
db_creator = importlib.import_module('cx_Oracle')
else:
raise Exception('unsupported database type ' +self.__db_type)
self.pool = PooledDB(
creator=db_creator,
mincached=0,
maxcached=6,
maxconnections=12,
blocking=True,
ping=1,
**config
)
def execute_query(self, sql, as_dict=True):
"""
查询语句
:paramsql:
:paramas_dict:
:return:
"""
conn =None
cur =None
try:
conn =self.pool.connection()
cur = conn.cursor()
cur.execute(sql)
rst = cur.fetchall()
if rst:
if as_dict:
fields = [tup[0]for tupin cur._cursor.description]
return [dict(zip(fields, row))for rowin rst]
return rst
return rst
except Exception as e:
print('sql:[{}]meet error'.format(sql))
print(e.args[-1])
return ()
finally:
if conn:
conn.close()
if cur:
cur.close()
def execute_manay(self, sql, *args):
"""
执行多条语句
:paramsql:
:paramdata:
:return:
"""
conn =None
cur =None
try:
conn =self.pool.connection()
cur = conn.cursor()
cur.executemany(sql, *args)
conn.commit()
return True
except Exception as e:
print('[{}]meet error'.format(sql))
print(e.args[-1])
conn.rollback()
return False
finally:
if conn:
conn.close()
if cur:
cur.close()
def execute(self, sql, *args):
"""
执行多条语句
:paramsql:
:paramdata:
:return:
"""
conn =None
cur =None
try:
conn =self.pool.connection()
cur = conn.cursor()
cur.execute(sql, *args)
conn.commit()
return True
except Exception as e:
print('[{}]meet error'.format(sql))
print(e.args[-1])
conn.rollback()
return False
finally:
if conn:
conn.close()
if cur:
cur.close()
def execute_ai(self, **sql):
sqls=sql['sql']
parms=sql['param']
conn =None
cur =None
try:
conn =self.pool.connection()
cur = conn.cursor()
for iin range(len(sqls)):
sql1 = sqls[i]
print(sqls[i], parms[i])
if str(sql1).find("update") >=0 or str(sql1).find("delete") >=0:
if parms[i]:
cur.execute(sql1, parms[i])
else:
cur.execute(sql1)
elif str(sql1).lower().find("insert") >=0:
cur.executemany(sql1, parms[i])
conn.commit()
return True
except Exception as e:
print('[{}]meet error'.format(sql))
print(e.args[-1])
conn.rollback()
return False
finally:
if conn:
conn.close()
if cur:
cur.close()
Code ThreadDb:
import vthread
from Db.DbPoolimport DbPool
sys.path.append(os.path.abspath('%s/..' % sys.path[0]))
from Db.dbconfigimport *
pool_1 = vthread.pool(5, gqueue=1)
class ThreadDb(DbPool):
"多线程并发MySQL插入数据"
def __init__(self,db_type,config):
super(ThreadDb, self).__init__(db_type,config)
self.data =self.getData()
self.mysql_delete()
def getData(self):
#产生测试数据
random_list =list(range(0,10000))
data = [('A'+str(i), 'B'+str(i),'C'+str(i),'D'+str(i))for iin random_list]
n=100
result = [data[i:i+n]for iin range(0, len(data),n)]
return result
def mysql_delete(self):
st = time.time()
con =self.pool.connection()
cur = con.cursor()
sql ="TRUNCATE TABLE baike.test"
cur.execute(sql)
con.commit()
cur.close()
con.close()
print("清空原数据.==>> 耗时:{}'s".format(round(time.time() - st, 3)))
@pool_1
def mysql_insert(self, content):
sql ="INSERT INTO baike.test(sku, fnsku, asin, shopid) VALUES(%s, %s, %s, %s)"
print(sql)
self.execute_manay(sql, content)
def mysql_update(self, content):
sql ="update baike.test set sku=%s,fnsku=%s"
print(sql)
self.execute(sql, content)
def task(self):
while self.data:
content =self.data.pop()
self.mysql_insert(content)
vthread.pool.wait(gqueue=1)
if __name__ =='__main__':
threadb=ThreadDb('mysql',dbconfig)
threadb.task()
5，小结
DbPool类是数据库池操作类，封装了增删改查的方法，里面有一个ai的方法主要作用是我们可以传一个sql的list和sql的一个参数list，可以在一个事务里完成多个sql的操作，不过这个只在 mysql测试过，其它的类型可以参考修改。
ThreadDb 类是一个多线程操作类，初始化里继承了DbPool类, 大家可以修改线程个数。
欢迎交流！
