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()

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

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()

def task():
 q = Queue(maxsize=10)  # 設(shè)定最大隊列數(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()

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()  # 事務(wù)回滾
print('SQL執(zhí)行有誤,原因:', e)
  finally:
cur.close()
con.close()
 def task(self):
  q = Queue(maxsize=10)  # 設(shè)定最大隊列數(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()