目錄
- 多線程(連接池)操作MySQL插入數(shù)據(jù)
- 1.主要模塊
- 2.創(chuàng)建連接池
- 3.數(shù)據(jù)預(yù)處理
- 4.線程任務(wù)
- 5.啟動(dòng)多線程
- 6.完整示例
- 7.思考/總結(jié)
多線程(連接池)操作MySQL插入數(shù)據(jù)
針對(duì)于此篇博客的收獲心得:
- 首先是可以構(gòu)建連接數(shù)據(jù)庫(kù)的連接池,這樣可以多開啟連接,同一時(shí)間連接不同的數(shù)據(jù)表進(jìn)行查詢,插入,為多線程進(jìn)行操作數(shù)據(jù)庫(kù)打基礎(chǔ)
- 多線程根據(jù)多連接的方式,需求中要完成多語言的入庫(kù)操作,我們可以啟用多線程對(duì)不同語言數(shù)據(jù)進(jìn)行并行操作
- 在插入過程中,一條一插入,比較浪費(fèi)時(shí)間,我們可以把數(shù)據(jù)進(jìn)行積累,積累到一定的條數(shù)的時(shí)候,執(zhí)行一條sql命令,一次性將多條數(shù)據(jù)插入到數(shù)據(jù)庫(kù)中,節(jié)省時(shí)間cur.executemany
1.主要模塊
DBUtils : 允許在多線程應(yīng)用和數(shù)據(jù)庫(kù)之間連接的模塊套件
Threading : 提供多線程功能
2.創(chuàng)建連接池
PooledDB 基本參數(shù):
- mincached : 最少的空閑連接數(shù),如果空閑連接數(shù)小于這個(gè)數(shù),Pool自動(dòng)創(chuàng)建新連接;
- maxcached : 最大的空閑連接數(shù),如果空閑連接數(shù)大于這個(gè)數(shù),Pool則關(guān)閉空閑連接;
- maxconnections : 最大的連接數(shù);
- blocking : 當(dāng)連接數(shù)達(dá)到最大的連接數(shù)時(shí),在請(qǐng)求連接的時(shí)候,如果這個(gè)值是True,請(qǐng)求連接的程序會(huì)一直等待,直到當(dāng)前連接數(shù)小于最大連接數(shù),如果這個(gè)值是False,會(huì)報(bào)錯(cuò);
def mysql_connection():
maxconnections = 15 # 最大連接數(shù)
pool = PooledDB(
pymysql,
maxconnections,
host='localhost',
user='root',
port=3306,
passwd='123456',
db='test_DB',
use_unicode=True)
return pool
# 使用方式
pool = mysql_connection()
con = pool.connection()
3.數(shù)據(jù)預(yù)處理
文件格式:txt
共準(zhǔn)備了四份虛擬數(shù)據(jù)以便測(cè)試,分別有10萬, 50萬, 100萬, 500萬行數(shù)據(jù)
MySQL表結(jié)構(gòu)如下圖:
數(shù)據(jù)處理思路 :
- 每一行一條記錄,每個(gè)字段間用制表符 “\t” 間隔開,字段帶有雙引號(hào);
- 讀取出來的數(shù)據(jù)類型是 Bytes ;
- 最終得到嵌套列表的格式,用于多線程循環(huán)每個(gè)任務(wù)每次處理10萬行數(shù)據(jù);
格式 : [ [(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("10w.txt", "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萬行數(shù)據(jù)為最小單位拆分成嵌套列表
result = [data[i:i + n] for i in range(0, len(data), n)]
print("10萬行數(shù)據(jù),耗時(shí):{}".format(round(time.time() - st, 3)))
# 10萬行數(shù)據(jù),耗時(shí):0.374
# 50萬行數(shù)據(jù),耗時(shí):1.848
# 100萬行數(shù)據(jù),耗時(shí):3.725
# 500萬行數(shù)據(jù),耗時(shí):18.493
4.線程任務(wù)
每調(diào)用一次插入函數(shù)就從連接池中取出一個(gè)鏈接操作,完成后關(guān)閉鏈接;
executemany 批量操作,減少 commit 次數(shù),提升效率;
def mysql_insert(*args):
con = pool.connection()
cur = con.cursor()
sql = "INSERT INTO test(sku,fnsku,asin,shopid) VALUES(%s, %s, %s, %s)"
try:
cur.executemany(sql, *args)
con.commit()
except Exception as e:
con.rollback() # 事務(wù)回滾
print('SQL執(zhí)行有誤,原因:', e)
finally:
cur.close()
con.close()
5.啟動(dòng)多線程
代碼思路 :
設(shè)定最大隊(duì)列數(shù),該值必須要小于連接池的最大連接數(shù),否則創(chuàng)建線程任務(wù)所需要的連接無法滿足,會(huì)報(bào)錯(cuò) : pymysql.err.OperationalError: (1040, ‘Too many connections')循環(huán)預(yù)處理好的列表數(shù)據(jù),添加隊(duì)列任務(wù)如果達(dá)到隊(duì)列最大值 或者 當(dāng)前任務(wù)是最后一個(gè),就開始多線程隊(duì)執(zhí)行隊(duì)列里的任務(wù),直到隊(duì)列為空;
def task():
q = Queue(maxsize=10) # 設(shè)定最大隊(duì)列數(shù)和線程數(shù)
# data : 預(yù)處理好的數(shù)據(jù)(嵌套列表)
while data:
content = data.pop()
t = threading.Thread(target=mysql_insert, args=(content,))
q.put(t)
if (q.full() == True) or (len(data)) == 0:
thread_list = []
while q.empty() == False:
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 ThreadInsert(object):
"多線程并發(fā)MySQL插入數(shù)據(jù)"
def __init__(self):
start_time = time.time()
self.pool = self.mysql_connection()
self.data = self.getData()
self.mysql_delete()
self.task()
print("========= 數(shù)據(jù)插入,共耗時(shí):{}'s =========".format(round(time.time() - start_time, 3)))
def mysql_connection(self):
maxconnections = 15 # 最大連接數(shù)
pool = PooledDB(
pymysql,
maxconnections,
host='localhost',
user='root',
port=3306,
passwd='123456',
db='test_DB',
use_unicode=True)
return pool
def getData(self):
st = time.time()
with open("10w.txt", "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萬行數(shù)據(jù)為最小單位拆分成嵌套列表
result = [data[i:i + n] for i in range(0, len(data), n)]
print("共獲取{}組數(shù)據(jù),每組{}個(gè)元素.==>> 耗時(shí):{}'s".format(len(result), n, round(time.time() - st, 3)))
return result
def mysql_delete(self):
st = time.time()
con = self.pool.connection()
cur = con.cursor()
sql = "TRUNCATE TABLE test"
cur.execute(sql)
con.commit()
cur.close()
con.close()
print("清空原數(shù)據(jù).==>> 耗時(shí):{}'s".format(round(time.time() - st, 3)))
def mysql_insert(self, *args):
con = self.pool.connection()
cur = con.cursor()
sql = "INSERT INTO test(sku, fnsku, asin, shopid) VALUES(%s, %s, %s, %s)"
try:
cur.executemany(sql, *args)
con.commit()
except Exception as e:
con.rollback() # 事務(wù)回滾
print('SQL執(zhí)行有誤,原因:', e)
finally:
cur.close()
con.close()
def task(self):
q = Queue(maxsize=10) # 設(shè)定最大隊(duì)列數(shù)和線程數(shù)
st = time.time()
while self.data:
content = self.data.pop()
t = threading.Thread(target=self.mysql_insert, args=(content,))
q.put(t)
if (q.full() == True) or (len(self.data)) == 0:
thread_list = []
while q.empty() == False:
t = q.get()
thread_list.append(t)
t.start()
for t in thread_list:
t.join()
print("數(shù)據(jù)插入完成.==>> 耗時(shí):{}'s".format(round(time.time() - st, 3)))
if __name__ == '__main__':
ThreadInsert()
插入數(shù)據(jù)對(duì)比
共獲取1組數(shù)據(jù),每組100000個(gè)元素.== >> 耗時(shí):0.374's
清空原數(shù)據(jù).== >> 耗時(shí):0.031's
數(shù)據(jù)插入完成.== >> 耗時(shí):2.499's
=============== 10w數(shù)據(jù)插入,共耗時(shí):3.092's ===============
共獲取5組數(shù)據(jù),每組100000個(gè)元素.== >> 耗時(shí):1.745's
清空原數(shù)據(jù).== >> 耗時(shí):0.0's
數(shù)據(jù)插入完成.== >> 耗時(shí):16.129's
=============== 50w數(shù)據(jù)插入,共耗時(shí):17.969's ===============
共獲取10組數(shù)據(jù),每組100000個(gè)元素.== >> 耗時(shí):3.858's
清空原數(shù)據(jù).== >> 耗時(shí):0.028's
數(shù)據(jù)插入完成.== >> 耗時(shí):41.269's
=============== 100w數(shù)據(jù)插入,共耗時(shí):45.257's ===============
共獲取50組數(shù)據(jù),每組100000個(gè)元素.== >> 耗時(shí):19.478's
清空原數(shù)據(jù).== >> 耗時(shí):0.016's
數(shù)據(jù)插入完成.== >> 耗時(shí):317.346's
=============== 500w數(shù)據(jù)插入,共耗時(shí):337.053's ===============
7.思考/總結(jié)
思考 :多線程+隊(duì)列的方式基本能滿足日常的工作需要,但是細(xì)想還是有不足;
例子中每次執(zhí)行10個(gè)線程任務(wù),在這10個(gè)任務(wù)執(zhí)行完后才能重新添加隊(duì)列任務(wù),這樣會(huì)造成隊(duì)列空閑.如剩余1個(gè)任務(wù)未完成,當(dāng)中空閑數(shù) 9,當(dāng)中的資源時(shí)間都浪費(fèi)了;
是否能一直保持隊(duì)列飽滿的狀態(tài),每完成一個(gè)任務(wù)就重新填充一個(gè).
到此這篇關(guān)于Python3 多線程(連接池)操作MySQL插入數(shù)據(jù)的文章就介紹到這了,更多相關(guān)Python3 多線程插入MySQL數(shù)據(jù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
您可能感興趣的文章:- Python3 操作 MySQL 插入一條數(shù)據(jù)并返回主鍵 id的實(shí)例
- python3實(shí)現(xiàn)往mysql中插入datetime類型的數(shù)據(jù)
- 使用python3 實(shí)現(xiàn)插入數(shù)據(jù)到mysql
- 解決python3插入mysql時(shí)內(nèi)容帶有引號(hào)的問題
- python3 pandas 讀取MySQL數(shù)據(jù)和插入的實(shí)例
- Python3.6-MySql中插入文件路徑,丟失反斜杠的解決方法