Python3 多線程(鏈接池)操做MySQL插入數據

1.主要模塊
DBUtils : 容許在多線程應用和數據庫之間鏈接的模塊套件
Threading : 提供多線程功能

2.建立鏈接池
PooledDB 基本參數：

mincached : 最少的空閒鏈接數，若是空閒鏈接數小於這個數，Pool自動建立新鏈接;
maxcached : 最大的空閒鏈接數，若是空閒鏈接數大於這個數，Pool則關閉空閒鏈接;
maxconnections : 最大的鏈接數;
blocking : 當鏈接數達到最大的鏈接數時，在請求鏈接的時候，若是這個值是True，請求鏈接的程序會一直等待，直到當前鏈接數小於最大鏈接數，若是這個值是False，會報錯;
CODE :

def mysql_connection():
    maxconnections = 15  # 最大鏈接數
    pool = PooledDB(
        pymysql,
        maxconnections,
        host='localhost',
        user='root',
        port=3306,
        passwd='123456',
        db='test_DB',
        use_unicode=True)
    return pool
# use
>> pool = mysql_connection()
>> con = pool.connection()

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),…], [], … ]
CODE :

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萬行數據爲最小單位拆分紅嵌套列表
    result = [data[i:i + n] for i in range(0, len(data), n)]
print("10萬行數據,耗時:{}".format(round(time.time() - st, 3)))

# out
>> 10萬行數據,耗時:0.374
>> 50萬行數據,耗時:1.848
>> 100萬行數據,耗時:3.725
>> 500萬行數據,耗時:18.493

4.線程任務

每調用一次插入函數就從鏈接池中取出一個連接操做,完成後關閉連接;
executemany 批量操做,減小 commit 次數,提高效率;

CODE :

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執行有誤,緣由:', e)
    finally:
        cur.close()
        con.close()

5.啓動多線程
代碼思路 :

設定最大隊列數,該值必需要小於鏈接池的最大鏈接數,不然建立線程任務所須要的鏈接沒法知足,會報錯 : pymysql.err.OperationalError: (1040, ‘Too many connections’)
循環預處理好的列表數據,添加隊列任務
若是達到隊列最大值 或者 當前任務是最後一個,就開始多線程隊執行隊列裏的任務,直到隊列爲空;
CODE :

def task():
    q = Queue(maxsize=10)  # 設定最大隊列數和線程數
    # data : 預處理好的數據(嵌套列表)
    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):
    "多線程併發MySQL插入數據"
    def __init__(self):
        start_time = time.time()
        self.pool = self.mysql_connection()
        self.data = self.getData()
        self.mysql_delete()
        self.task()
        print("========= 數據插入,共耗時:{}'s =========".format(round(time.time() - start_time, 3)))

    def mysql_connection(self):
        maxconnections = 15  # 最大鏈接數
        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萬行數據爲最小單位拆分紅嵌套列表
        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(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("清空原數據.==>> 耗時:{}'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執行有誤,緣由:', e)
        finally:
            cur.close()
            con.close()

    def task(self):
        q = Queue(maxsize=10)  # 設定最大隊列數和線程數
        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("數據插入完成.==>> 耗時:{}'s".format(round(time.time() - st, 3)))


if __name__ == '__main__':
    ThreadInsert()

插入數據對比

共獲取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,當中的資源時間都浪費了;是否能一直保持隊列飽滿的狀態,每完成一個任務就從新填充一個.總結 :野生猿一枚,代碼很粗糙,若是錯誤請評論指正.