溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹Python3中多線程操作MySQL插入數(shù)據(jù)的方法，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！
針對于此篇博客的收獲心得：
首先是可以構建連接數(shù)據(jù)庫的連接池，這樣可以多開啟連接，同一時間連接不同的數(shù)據(jù)表進行查詢，插入，為多線程進行操作數(shù)據(jù)庫打基礎
多線程根據(jù)多連接的方式，需求中要完成多語言的入庫操作，我們可以啟用多線程對不同語言數(shù)據(jù)進行并行操作
在插入過程中，一條一插入，比較浪費時間，我們可以把數(shù)據(jù)進行積累，積累到一定的條數(shù)的時候，執(zhí)行一條sql命令，一次性將多條數(shù)據(jù)插入到數(shù)據(jù)庫中，節(jié)省時間cur.executemany
DBUtils : 允許在多線程應用和數(shù)據(jù)庫之間連接的模塊套件
Threading : 提供多線程功能
PooledDB 基本參數(shù)：
mincached : 最少的空閑連接數(shù)，如果空閑連接數(shù)小于這個數(shù)，Pool自動創(chuàng)建新連接;
maxcached : 最大的空閑連接數(shù)，如果空閑連接數(shù)大于這個數(shù)，Pool則關閉空閑連接;
maxconnections : 最大的連接數(shù);
blocking : 當連接數(shù)達到最大的連接數(shù)時，在請求連接的時候，如果這個值是True，請求連接的程序會一直等待，直到當前連接數(shù)小于最大連接數(shù)，如果這個值是False，會報錯;
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()
文件格式:txt
共準備了四份虛擬數(shù)據(jù)以便測試,分別有10萬, 50萬, 100萬, 500萬行數(shù)據(jù)
MySQL表結構如下圖:
數(shù)據(jù)處理思路 :
每一行一條記錄,每個字段間用制表符 “\t” 間隔開,字段帶有雙引號;
讀取出來的數(shù)據(jù)類型是 Bytes ;
最終得到嵌套列表的格式,用于多線程循環(huán)每個任務每次處理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ù),耗時:{}".format(round(time.time() - st, 3)))
# 10萬行數(shù)據(jù),耗時:0.374
# 50萬行數(shù)據(jù),耗時:1.848
# 100萬行數(shù)據(jù),耗時:3.725
# 500萬行數(shù)據(jù),耗時:18.493每調用一次插入函數(shù)就從連接池中取出一個鏈接操作,完成后關閉鏈接;
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() # 事務回滾
print('SQL執(zhí)行有誤,原因:', e)
finally:
cur.close()
con.close()代碼思路 :
設定最大隊列數(shù),該值必須要小于連接池的最大連接數(shù),否則創(chuàng)建線程任務所需要的連接無法滿足,會報錯 : pymysql.err.OperationalError: (1040, ‘Too many connections')循環(huán)預處理好的列表數(shù)據(jù),添加隊列任務如果達到隊列最大值 或者 當前任務是最后一個,就開始多線程隊執(zhí)行隊列里的任務,直到隊列為空;
def task(): q = Queue(maxsize=10) # 設定最大隊列數(shù)和線程數(shù) # data : 預處理好的數(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()
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ù)插入,共耗時:{}'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ù),每組{}個元素.==>> 耗時:{}'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ù).==>> 耗時:{}'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() # 事務回滾
print('SQL執(zhí)行有誤,原因:', e)
finally:
cur.close()
con.close()
def task(self):
q = Queue(maxsize=10) # 設定最大隊列數(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ù)插入完成.==>> 耗時:{}'s".format(round(time.time() - st, 3)))
if __name__ == '__main__':
ThreadInsert()插入數(shù)據(jù)對比
共獲取1組數(shù)據(jù),每組100000個元素.== >> 耗時:0.374's
清空原數(shù)據(jù).== >> 耗時:0.031's
數(shù)據(jù)插入完成.== >> 耗時:2.499's
=============== 10w數(shù)據(jù)插入,共耗時:3.092's ===============
共獲取5組數(shù)據(jù),每組100000個元素.== >> 耗時:1.745's
清空原數(shù)據(jù).== >> 耗時:0.0's
數(shù)據(jù)插入完成.== >> 耗時:16.129's
=============== 50w數(shù)據(jù)插入,共耗時:17.969's ===============
共獲取10組數(shù)據(jù),每組100000個元素.== >> 耗時:3.858's
清空原數(shù)據(jù).== >> 耗時:0.028's
數(shù)據(jù)插入完成.== >> 耗時:41.269's
=============== 100w數(shù)據(jù)插入,共耗時:45.257's ===============
共獲取50組數(shù)據(jù),每組100000個元素.== >> 耗時:19.478's
清空原數(shù)據(jù).== >> 耗時:0.016's
數(shù)據(jù)插入完成.== >> 耗時:317.346's
=============== 500w數(shù)據(jù)插入,共耗時:337.053's ===============
思考 :多線程+隊列的方式基本能滿足日常的工作需要,但是細想還是有不足;
例子中每次執(zhí)行10個線程任務,在這10個任務執(zhí)行完后才能重新添加隊列任務,這樣會造成隊列空閑.如剩余1個任務未完成,當中空閑數(shù) 9,當中的資源時間都浪費了;
是否能一直保持隊列飽滿的狀態(tài),每完成一個任務就重新填充一個.
以上是“Python3中多線程操作MySQL插入數(shù)據(jù)的方法”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注億速云行業(yè)資訊頻道！
免責聲明：本站發(fā)布的內容（圖片、視頻和文字）以原創(chuàng)、轉載和分享為主，文章觀點不代表本網(wǎng)站立場，如果涉及侵權請聯(lián)系站長郵箱：is@yisu.com進行舉報，并提供相關證據(jù)，一經(jīng)查實，將立刻刪除涉嫌侵權內容。
