python 多線程與多進(jìn)程效率測試

更新時間：2021年10月26日 10:45:00 作者：Silent丿丶黑羽

這篇文章主要介紹了python 多線程與多進(jìn)程效率測試，在Python中，計算密集型任務(wù)適用于多進(jìn)程，IO密集型任務(wù)適用于多線程、接下來看看文章得實(shí)例吧,需要的朋友可以參考一下喲

1、概述

在Python中，計算密集型任務(wù)適用于多進(jìn)程，IO密集型任務(wù)適用于多線程

正常來講，多線程要比多進(jìn)程效率更高，因為進(jìn)程間的切換需要的資源和開銷更大，而線程相對更小，但是我們使用的Python大多數(shù)的解釋器是Cpython，眾所周知Cpython有個GIL鎖，導(dǎo)致執(zhí)行計算密集型任務(wù)時多線程實(shí)際只能是單線程，而且由于線程之間切換的開銷導(dǎo)致多線程往往比實(shí)際的單線程還要慢，所以在 python 中計算密集型任務(wù)通常使用多進(jìn)程，因為各個進(jìn)程有各自獨(dú)立的GIL，互不干擾。

而在IO密集型任務(wù)中，CPU時常處于等待狀態(tài)，操作系統(tǒng)需要頻繁與外界環(huán)境進(jìn)行交互，如讀寫文件，在網(wǎng)絡(luò)間通信等。在這期間GIL會被釋放，因而就可以使用真正的多線程。

上面都是理論，接下來實(shí)戰(zhàn)看看實(shí)際效果是否符合理論

2、代碼練習(xí)

"""多線程多進(jìn)程模擬執(zhí)行效率"""


from multiprocessing import Pool
from threading import Thread
import time, math


def simulation_IO(a):
    """模擬IO操作"""
    time.sleep(3)


def simulation_compute(a):
    """模擬計算密集型任務(wù)"""
    for i in range(int(1e7)):
        math.sin(40) + math.cos(40)
    return


def normal_func(func):
    """普通方法執(zhí)行效率"""
    for i in range(6):
        func(i)
    return


def mp(func):
    """進(jìn)程池中的map方法"""
    with Pool(processes=6) as p:
        res = p.map(func, list(range(6)))
    return


def asy(func):
    """進(jìn)程池中的異步執(zhí)行"""
    with Pool(processes=6) as p:
        result = []
        for j in range(6):
            a = p.apply_async(func, args=(j, ))
            result.append(a)
        res = [j.get() for j in result]


def thread(func):
    """多線程方法"""
    threads = []
    for j in range(6):
        t = Thread(target=func, args=(j, ))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()


def showtime(f, func, name):
    """
    計算并展示函數(shù)的運(yùn)行時間
    :param f: 多進(jìn)程和多線程的方法
    :param func: 多進(jìn)程和多線程方法中需要傳入的函數(shù)
    :param name: 方法的名字
    :return:
    """
    start_time = time.time()
    f(func)
    print(f"{name} time: {time.time() - start_time:.4f}s")


def main(func):
    """
    運(yùn)行程序的主函數(shù)
    :param func: 傳入需要計算時間的函數(shù)名
    """
    showtime(normal_func, func, "normal")
    print()
    print("------ 多進(jìn)程 ------")
    showtime(mp, func, "map")
    showtime(asy, func, "async")
    print()
    print("----- 多線程 -----")
    showtime(thread, func, "thread")


if __name__ == "__main__":
    print("------------ 計算密集型 ------------")
    func = simulation_compute
    main(func)
    print()
    print()
    print()
    print("------------ IO 密集型 ------------")
    func = simulation_IO
    main(func)