python 多线程与多进程效率测试_Python

python 多线程与多进程效率测试

2022-02-18 00:27Silent丿丶黑羽 Python

这篇文章主要介绍了python 多线程与多进程效率测试，在Python中，计算密集型任务适用于多进程，IO密集型任务适用于多线程、接下来看看文章得实例吧,需要的朋友可以参考一下哟

1、概述

在Python中，计算密集型任务适用于多进程，IO密集型任务适用于多线程

正常来讲，多线程要比多进程效率更高，因为进程间的切换需要的资源和开销更大，而线程相对更小，但是我们使用的Python大多数的解释器是Cpython，众所周知Cpython有个GIL锁，导致执行计算密集型任务时多线程实际只能是单线程，而且由于线程之间切换的开销导致多线程往往比实际的单线程还要慢，所以在 python 中计算密集型任务通常使用多进程，因为各个进程有各自独立的GIL，互不干扰。

而在IO密集型任务中，CPU时常处于等待状态，操作系统需要频繁与外界环境进行交互，如读写文件，在网络间通信等。在这期间GIL会被释放，因而就可以使用真正的多线程。

上面都是理论，接下来实战看看实际效果是否符合理论

2、代码练习

				?

									"""多线程多进程模拟执行效率"""

									from multiprocessing import Pool

									from threading import Thread

									import time, math

									def simulation_IO(a):

									    """模拟IO操作"""

									    time.sleep(3)

									def simulation_compute(a):

									    """模拟计算密集型任务"""

									    for i in range(int(1e7)):

									        math.sin(40) + math.cos(40)

									    return

									def normal_func(func):

									    """普通方法执行效率"""

									    for i in range(6):

									        func(i)

									    return

									def mp(func):

									    """进程池中的map方法"""

									    with Pool(processes=6) as p:

									        res = p.map(func, list(range(6)))

									    return

									def asy(func):

									    """进程池中的异步执行"""

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

									    """

									    计算并展示函数的运行时间

									    :param f: 多进程和多线程的方法

									    :param func: 多进程和多线程方法中需要传入的函数

									    :param name: 方法的名字

									    :return:

									    """

									    start_time = time.time()

									    f(func)

									    print(f"{name} time: {time.time() - start_time:.4f}s")

									def main(func):

									    """

									    运行程序的主函数

									    :param func: 传入需要计算时间的函数名

									    """

									    showtime(normal_func, func, "normal")

									    print()

									    print("------ 多进程 ------")

									    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)