Python3多线程（连接池）操作MySQL插⼊数据
⽬录
多线程(连接池)操作MySQL插⼊数据
1.主要模块
2.创建连接池
3.数据预处理
4.线程任务
5.启动多线程
6.完整⽰例
7.思考/总结
多线程(连接池)操作MySQL插⼊数据
针对于此篇博客的收获⼼得：
⾸先是可以构建连接数据库的连接池，这样可以多开启连接，同⼀时间连接不同的数据表进⾏查询，插⼊，为多线程进⾏操作数据库打基础
多线程根据多连接的⽅式，需求中要完成多语⾔的⼊库操作，我们可以启⽤多线程对不同语⾔数据进⾏并⾏操作
在插⼊过程中，⼀条⼀插⼊，⽐较浪费时间，我们可以把数据进⾏积累，积累到⼀定的条数的时候，执⾏⼀条sql命令，⼀次性将多条数据插⼊到数据库中，节省时间utemany
1.主要模块
DBUtils : 允许在多线程应⽤和数据库之间连接的模块套件
Threading : 提供多线程功能
2.创建连接池
yodaoPooledDB 基本参数：
mincached : 最少的空闲连接数，如果空闲连接数⼩于这个数，Pool⾃动创建新连接;
maxcached : 最⼤的空闲连接数，如果空闲连接数⼤于这个数，Pool则关闭空闲连接;
maxconnections : 最⼤的连接数;
blocking : 当连接数达到最⼤的连接数时，在请求连接的时候，如果这个值是True，请求连接的程序会⼀直等待，直到当前连接数⼩于最⼤连接数，如果这个值是Fal，会报错;
def mysql_connection():
maxconnections = 15 # 最⼤连接数band
pool = PooledDB(
pymysql,
maxconnections,
host='localhost',
ur='root',
port=3306,
passwd='123456',
db='test_DB',
u_unicode=True)
return pool
# 使⽤⽅式company的复数
pool = mysql_connection()
con = tion()
3.数据预处理
⽂件格式:txt
共准备了四份虚拟数据以便测试，分别有10万, 50万, 100万, 500万⾏数据
MySQL表结构如下图:
数据处理思路 :
每⼀⾏⼀条记录，每个字段间⽤制表符 “\t” 间隔开，字段带有双引号;
读取出来的数据类型是 Bytes ;
最终得到嵌套列表的格式,⽤于多线程循环每个任务每次处理10万⾏数据;
格式 : [ [(A,B,C,D), (A,B,C,D),(A,B,C,D),…], [(A,B,C,D), (A,B,C,D),(A,B,C,D),…], [], … ]
import re
import time
st = time.time()
with open("", "rb") as f:
data = []
for line in f:
line = re.sub("\s", "", str(line, encoding="utf-8"))
line = tuple(line[1:-1].split("\"\""))灰度值
www asian4you com
data.append(line)
n = 100000 # 按每10万⾏数据为最⼩单位拆分成嵌套列表
result = [data[i:i + n] for i in range(0, len(data), n)]新托福考试冲刺试题
print("10万⾏数据，耗时:{}".format(round(time.time() - st, 3)))
# 10万⾏数据，耗时:0.374
# 50万⾏数据，耗时:1.848
# 100万⾏数据，耗时:3.725
# 500万⾏数据，耗时:18.493
4.线程任务
plz每调⽤⼀次插⼊函数就从连接池中取出⼀个链接操作，完成后关闭链接;
executemany 批量操作，减少 commit 次数，提升效率;
def mysql_inrt(*args):
con = tion()
cur = con.cursor()
sql = "INSERT INTO test(sku,fnsku,asin,shopid) VALUES(%s, %s, %s, %s)"
try:
except Exception as e:
print('SQL执⾏有误，原因:', e)
韩语班培训finally:
cur.clo()
con.clo()
5.启动多线程
代码思路 :
设定最⼤队列数，该值必须要⼩于连接池的最⼤连接数，否则创建线程任务所需要的连接⽆法满⾜，会报错 :
def task():
matter是什么意思啊q = Queue(maxsize=10) # 设定最⼤队列数和线程数
# data : 预处理好的数据(嵌套列表)
while data:
content = data.pop()
t = threading.Thread(target=mysql_inrt, args=(content,))
q.put(t)
if (q.full() == True) or (len(data)) == 0:
thread_list = []
pty() == Fal:
t = q.get()
thread_list.append(t)
t.start()
for t in thread_list:
t.join()
6.完整⽰例
import pymysql
import threading
import re
import time
from queue import Queue
from DBUtils.PooledDB import PooledDB
class ThreadInrt(object):
"多线程并发MySQL插⼊数据"
def __init__(lf):
start_time = time.time()
lf.pool = lf.mysql_connection()
lf.data = lf.getData()
lf.task()
print("========= 数据插⼊，共耗时:{}'s =========".format(round(time.time() - start_time, 3))) def mysql_connection(lf):
maxconnections = 15 # 最⼤连接数
pool = PooledDB(
pymysql,
maxconnections,
host='localhost',
ur='root',
port=3306,
passwd='123456',
db='test_DB',
u_unicode=True)
return pool
def getData(lf):
st = time.time()
with open("", "rb") as f:
data = []
for line in f:
line = re.sub("\s", "", str(line, encoding="utf-8"))
line = tuple(line[1:-1].split("\"\""))
data.append(line)
n = 100000 # 按每10万⾏数据为最⼩单位拆分成嵌套列表
result = [data[i:i + n] for i in range(0, len(data), n)]
print("共获取{}组数据，每组{}个元素.==>> 耗时:{}'s".format(len(result), n, round(time.time() - st, 3))) return result
def mysql_delete(lf):
st = time.time()
con = tion()
cur = con.cursor()
sql = "TRUNCATE TABLE test"
cur.clo()
con.clo()
print("清空原数据.==>> 耗时:{}'s".format(round(time.time() - st, 3)))
def mysql_inrt(lf, *args):
con = tion()
cur = con.cursor()
sql = "INSERT INTO test(sku, fnsku, asin, shopid) VALUES(%s, %s, %s, %s)"
try:
except Exception as e:
print('SQL执⾏有误，原因:', e)
finally:
cur.clo()
postgraduate
con.clo()
def task(lf):
q = Queue(maxsize=10) # 设定最⼤队列数和线程数
st = time.time()
while lf.data:
content = lf.data.pop()
t = threading.Thread(sql_inrt, args=(content,))
q.put(t)
if (q.full() == True) or (len(lf.data)) == 0:
thread_list = []
pty() == Fal:
t = q.get()
thread_list.append(t)
t.start()
for t in thread_list:
t.join()
print("数据插⼊完成.==>> 耗时:{}'s".format(round(time.time() - st, 3)))
if __name__ == '__main__':
ThreadInrt()
插⼊数据对⽐
共获取1组数据，每组100000个元素.== >> 耗时:0.374's
清空原数据.== >> 耗时:0.031's
数据插⼊完成.== >> 耗时:2.499's
=============== 10w数据插⼊，共耗时:3.092's ===============
共获取5组数据，每组100000个元素.== >> 耗时:1.745's
清空原数据.== >> 耗时:0.0's
数据插⼊完成.== >> 耗时:16.129's
=============== 50w数据插⼊，共耗时:17.969's ===============
共获取10组数据，每组100000个元素.== >> 耗时:3.858's
清空原数据.== >> 耗时:0.028's
数据插⼊完成.== >> 耗时:41.269's
=============== 100w数据插⼊，共耗时:45.257's ===============
共获取50组数据，每组100000个元素.== >> 耗时:19.478's
清空原数据.== >> 耗时:0.016's
数据插⼊完成.== >> 耗时:317.346's
=============== 500w数据插⼊，共耗时:337.053's ===============
7.思考/总结
思考 ：多线程+队列的⽅式基本能满⾜⽇常的⼯作需要，但是细想还是有不⾜;
例⼦中每次执⾏10个线程任务，在这10个任务执⾏完后才能重新添加队列任务，这样会造成队列空闲.如剩余1个任务未完成，当中空闲数 9，当中的资源时间都浪费了;
是否能⼀直保持队列饱满的状态，每完成⼀个任务就重新填充⼀个.
到此这篇关于Python3 多线程(连接池)操作MySQL插⼊数据的⽂章就介绍到这了，更多相关Python3 多线程插⼊MySQL数据内容请搜索以前的⽂章或继续浏览下⾯的相关⽂章希望⼤家以后多多⽀持！
