python3教程：用concurrent执行多进程任务的方法

Posted 2022-06-25 Python热爱者

随着计算机技术的发展，诸如GPU和超算平台等越来越发达，这些技术的本质其实并没有带来算法上的革新，之所以能够提升计算的速度和规模，很大程度上是因为分布式和并行计算的优势。这里我们介绍一个简单的python自带的多进程的代码实现，使用的是concurrent这个工具，同时我们也会介绍如何更好的配置多进程的资源。

concurrent使用示例

concurrent是python自带的一个多进程实现仓库，不需要额外的安装。这里我们先介绍一个没有多进程的示例代码：

import time

def sleep(seconds):
    time.sleep(seconds)

if __name__ == __main__:
    times = [1] * 10
    time0 = time.time()
    for t in times:
        sleep(t)
    time1 = time.time()
    print (The time cost is: s.format(time1 - time0))

这个代码的功能其实就是休眠10s的时间，也比较容易，我们可以看一下这个代码的执行结果：

[dechin@dechin-manjaro concurrent]$ python3 sleep.py 
The time cost is: 10.014754295349121s

在我们统计时间的时候，发现会比10s的时间多出来一些，这部分时间不仅仅包含这个python程序执行的时间，跟时间的统计方式也有一定的关系，一般我们可以忽略这部分的gap时间。

我们假定上面这个程序中的sleep函数功能不是休眠1s的时间，而是执行一个耗时为1s的计算任务，而我们有很多的cpu，希望能够加速这个计算的过程，这个时候我们就需要用到多进程的技术，下面是修改为多进程之后的代码：

import concurrent.futures
import time

def sleep(seconds):
    time.sleep(seconds)

if __name__ == __main__:
    times = [1] * 10
    time0 = time.time()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        executor.map(sleep, times)
    time1 = time.time()
    print (The time cost is: s.format(time1 - time0))

整个修改的方式也非常容易，就是把原代码中的for循环修改为concurrent的执行语句，让我们一起来看看执行的结果：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py 
The time cost is: 2.0304219722747803s

从结果上我们可以看到，运行的时间从10s降低到了2s。这说明我们的多进程任务带来了非常显著的优化效果，至于为什么优化之后是2s而不是3s或者1s，这个问题将在下一个章节中进行介绍。

多进程的最佳配置

使用多进程的方案，能有多大的加速效果，很大程度上是依赖于硬件的。理论上来说，如果有n个cpu核，我们就可以实现n倍的加速。但是大部分情况下会受限于算法或者任务本身，跟n倍的加速之间有一定的gap。首先让我们用ipython来看看本地电脑上有多少个cpu：

[dechin@dechin-manjaro concurrent]$ ipython
Python 3.8.5 (default, Sep  4 2020, 07:30:14) 
Type copyright, credits or license for more information
IPython 7.19.0 -- An enhanced Interactive Python. Type ? for help.

In [1]: import psutil

In [2]: psutil.cpu_count(logical=False)
Out[2]: 4

In [3]: psutil.cpu_count(logical=True)
Out[3]: 8

这里我们使用的是psutil这个库，而不是常用的os或者multiprocessing，这是因为可以更好的区分逻辑核与物理核。我们本地的电脑上有4个物理核，每个物理核实际上对应于2个逻辑核，因此一共是有8个逻辑核。也就是说，理论上我们最大可以加速8倍的算法执行时间。让我们通过配置和修改一些参数来测试验证一下：

import concurrent.futures
import time
import sys

def sleep(seconds):
    time.sleep(seconds)

if __name__ == __main__:
    if sys.argv[1] == -t:
        times = [1] * int(sys.argv[2]) # 获取命令行的时间输入参数
    time0 = time.time()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        executor.map(sleep, times)
    time1 = time.time()
    print (The time cost is: s.format(time1 - time0))

在这个示例中，为了方便调整，我们把总的休眠时间定为命令行的输入参数，使用的是sys.argv这个函数来获取，注意获取到的参数是字符串格式的。这串代码的执行方法和执行结果如下所示：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py -t 16
The time cost is: 2.0304934978485107s

在上面的执行结果中，我们发现原本需要16s的休眠任务，在多进程场景下被加速到了2s，刚好符合我们对逻辑核的加速倍数的预期。但是前面提到，能否加速逻辑核的倍数，跟任务本身的算法也有关系。比如在本用例中，如果算法分配的结果就是需要17个子算法来进行实现，那么在每个逻辑核上面执行了2次的休眠任务之后，又有一个逻辑核需要再执行1次休眠任务，而此时其他的逻辑核都需要等待这个执行任务的逻辑核的任务结束。具体的执行结果如下所示：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py -t 17
The time cost is: 3.0313029289245605s

这个结果也验证了我们此前的预想，因为16s的任务执行需要2s的时间，而执行完16s的任务之后，又需要等待剩下的一个1s时长的任务执行完毕，总耗时3s。这里我们没有配置max_worker的情况下，会按照系统中最高的逻辑核数来进行多进程的任务分配，但是在实际场景中我们需要考虑多种因素的限制，如内存和进程数的均衡配置（在大内存任务中，如果进程全开，有可能导致内存不足的问题）。只有在满足所有系统约束条件的情况下，才能发挥硬件最好的性能。在下面的代码中我们将给出如何配置执行任务的核数的方案：

import concurrent.futures
import time
import sys

def sleep(seconds):
    time.sleep(seconds)

if __name__ == __main__:
    if sys.argv[1] == -t:
        times = [1] * int(sys.argv[2])
    time0 = time.time()
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        executor.map(sleep, times)
    time1 = time.time()
    print (The time cost is: s.以上是关于python3教程：用concurrent执行多进程任务的方法的主要内容，如果未能解决你的问题，请参考以下文章
 
 python 期物
 python3执行javaScript脚本，js2py教程
 swoft2 小白教程系列-进程篇
 多线程和多进程效率的对比
 Python3模块concurrent.futures模块，线程池进程池
 如何在 Python3 中检测 concurrent.futures 中的异常？