背景

??測試項目中，碰到一個場景，大體是需要驗證多個表中的數據正確性。8000W的用戶量，所以寫一個腳本來抽查其中一部分。
??初步使用python寫單線程腳本，跑1W數據，大約需要十幾分鐘。可見效率之低下。于是考慮使用Multiprocessing來加快執(zhí)行速度，以及對比下效果。
??本文不做模塊科普，只從實戰(zhàn)角度講起，教你如何快速把單線程變成多進程方式來加快執(zhí)行速度。

Multiprocessing 模塊使用

多進程使用要點概覽：

• 創(chuàng)建Pool，進程池
• map,開始執(zhí)行
• close， 關閉進程池
• join， 等待子進程執(zhí)行完畢

代碼demo

單線程版

import pymysql
import time
# 數據庫配置
db_config = {
  "host": "172.21.0.9",
  "port": 3306,
  "user": "root",
  "passwd": "ni cai bu dao",
  "db": "database_name"
}

# 數據庫查詢封裝
class DB(object):
    def __init__(self):
        db = pymysql.Connect(**db_config)
        self.cursor = db.cursor()
        
    # 查詢用戶最近一筆流水時間
    def query_user_flow(self, user_id):
        ext = int(user_id) % 100
        sql = 'select ctime from database_name.user_flow_{} where user_id={} order by id desc limit 1;'.format(ext, user_id)
        self.cursor.execute(sql)
        res = self.cursor.fetchone()
        return res


if __name__ == '__main__':
    start_time = time.time()
    # 讀取用戶id
    f = open('./user_credit_bk.txt', 'r').readlines()
    # 寫入流水時間到新文件
    nf = open('./user_time.txt', 'w')
    flow = DB()
    # 初始化一個字典
    user_dict = {}
    # 遍歷userid
    for line in f:
        user = line.split('\t')[0]
        atime = flow.query_user_flow(user)
        # 將userid和最近流水時間設為字典
        user_dict.setdefault(user, atime[0])
    # 按照流水時間倒序排序
    new_list = sorted(user_dict.items(), key=lambda item: item[1], reverse=True)
    # 遍歷列表，寫入新文件中
    for item in new_list:
        nf.write('{time}\t{user}\n'.format(time=item[1], user=item[0]))
    nf.close()
    end_time = time.time()
    print('程序總共執(zhí)行了{}s'.format(int(end_time - start_time)))

執(zhí)行結果：

程序總共執(zhí)行了144s

多進程版

import pymysql
import time
from multiprocessing import Manager, Pool
# 數據庫配置
db_config = {
  "host": "172.21.0.9",
  "port": 3306,
  "user": "root",
  "passwd": "ni cai bu dao",
  "db": "database_name"
}


# 數據庫查詢封裝
class DB(object):
    def __init__(self):
        db = pymysql.Connect(**db_config)
        self.cursor = db.cursor()
        
    # 查詢用戶最近一筆流水時間
    def query_user_flow(self, user_id):
        ext = int(user_id) % 100
        sql = 'select ctime from database_name.user_flow_{} where user_id={} order by id desc limit 1;'.format(ext, user_id)
        self.cursor.execute(sql)
        res = self.cursor.fetchone()
        return res
        
        
def user_time(user_info):
    flow = DB()
    user = user_info.split('\t')[0]
    atime = flow.query_user_flow(user)
    user_dict.setdefault(user, atime[0])


if __name__ == '__main__':
    # 開始時間
    start_time = time.time()
    # 讀取用戶數據
    user_infos = open('./user_credit_bk.txt', 'r').readlines()
    # 寫入流水時間到新文件
    nf = open('./user_time.txt', 'w')
    # 初始化一個字典(進程共享變量)
    user_dict = Manager().dict()
    # 創(chuàng)建10個進程
    with Pool(10) as pool:
        pool.map(user_time, user_infos)
        pool.close()
        pool.join()
    # 按照流水時間倒序排序
    new_list = sorted(user_dict.items(), key=lambda item: item[1], reverse=True)
    # 遍歷列表，寫入新文件中
    for item in new_list:
        nf.write('{time}\t{user}\n'.format(time=item[1], user=item[0]))
    nf.close()
    end_time = time.time()
    print('程序總共執(zhí)行了{}s'.format(int(end_time-start_time)))

執(zhí)行結果：

程序總共執(zhí)行了30s

總結

??可見多進程比單進程的快了將近5倍。
??其實Multiprocessing里還有非常多的用法，比如消息隊列，進程鎖等等。有興趣的同學可以一起來討論。