Python3 多線程(連接池)操作MySQL插入數(shù)據(jù)

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()  # 事務回滾
        print('SQL執(zhí)行有誤,原因:', e)
    finally:
        cur.close()
        con.close()

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