在 Python 脚本中同时使用多处理和多线程来加快执行速度

Posted 2023-02-16

【中文标题】在 Python 脚本中同时使用多处理和多线程来加快执行速度

【英文标题】：Using both multiprocessing and multithreading in a Python script to speed up execution

【发布时间】：2021-01-06 15:07:39

【问题描述】：

我有以下子网范围：10.106.44.0/24 - 10.106.71.0/24。我正在编写一个 Python 脚本来 ping 所有子网中的每个 IP。为了加速这个脚本，我尝试同时使用多处理和多线程。我正在为每个子网创建一个新进程并创建一个新线程来 ping 该子网中的每个主机。我想问两个问题：

这是解决此问题的最佳方法吗？ 如果是，我将如何实施？

【问题讨论】：

由于你只是 ping（主要是 IO）而不做任何长时间的计算，你可以只使用线程池。

但是如果我在我的服务器上的所有 12 个内核上运行它，它不会在 1/12 的时间内运行吗？

您可以将池大小设置为 12（作为您拥有的核心数）启动进程/线程有很大的开销。通过使用线程池，您只需初始化线程一次，它们就会被重用。

Python multiprocessing 也支持创建进程池（如果真的需要新进程的话）

多处理带来了有趣的设计挑战和成本 - 进程之间发送的所有数据都必须经过腌制/取消腌制，并且没有共享信息的全局变量。这里有很多问题，比如‘为什么我的多处理代码在 12 个内核上运行速度不能快 12 倍？甚至是‘为什么我的多处理代码比非多处理代码慢。线程更容易使用，并且对于 I/O 密集型工作非常有效。

【参考方案1】：

我会首先尝试使用线程。您可以尝试创建一个线程池，其大小是您必须执行的 ping 总数，但最终我相信这不会比使用等于您拥有的 CPU 内核数的线程池大小更好（下面的解释） .以下是使用线程和多处理两种方式的比较：

ThreadPoolExecutor（255 个线程）

from concurrent.futures import ThreadPoolExecutor
import os
import platform
import subprocess
import time

def ping_ip(ip_address):
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    try:
        output = subprocess.check_output(f"ping param 1 ip_address", shell=True, universal_newlines=True)
        if 'unreachable' in output:
            return False
        else:
            return True
    except Exception:
            return False


def main():
    t1 = time.time()
    ip_addresses = ['192.168.1.154'] * 255
    #with ThreadPoolExecutor(os.cpu_count())) as executor: # uses number of CPU cores
    with ThreadPoolExecutor(len(ip_addresses)) as executor:
        results = list(executor.map(ping_ip, ip_addresses))
        #print(results)
        print(time.time() - t1)

if __name__ == '__main__':
    main()

打印：

2.049474000930786

您可以尝试使用 更少 个线程（ThreadPoolExecutor 构造函数的max_workers 参数）。见：concurrent.futures

我发现运行 8 个线程（这是我拥有的核心数量）的效果也差不多（时间：2.2745485305786133）。我相信这样做的原因是，尽管 ping 是一项与 I/O 相关的任务，但对子进程的调用必须在内部创建一个使用大量 CPU 的新进程，因此并发性在某种程度上受到处理器限制。

ProcessPoolExecutor（8 核）

from concurrent.futures import ProcessPoolExecutor
import os
import platform
import subprocess
import time

def ping_ip(ip_address):
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    try:
        output = subprocess.check_output(f"ping param 1 ip_address", shell=True, universal_newlines=True)
        if 'unreachable' in output:
            return False
        else:
            return True
    except Exception:
            return False


def main():
    t1 = time.time()
    ip_addresses = ['192.168.1.154'] * 255
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(ping_ip, ip_addresses))
        #print(results)
        print(time.time() - t1)

if __name__ == '__main__':
    main()

打印：

2.509838819503784

请注意，在我的 Linux 系统上，您必须是超级用户才能发出 ping 命令。