python 多进程及并行计算： multiprocessing总结 & joblib.Parallel函数

2024-07-13 20:21| 来源: 网络整理| 查看: 265

一、背景由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。 Python提供了非常好用的多进程包multiprocessing，只需要定义一个函数，Python会完成其他所有事情。借助这个包，可以轻松完成从单进程到并发执行的转换。 multiprocessing支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。二、 multiprocessing包介绍

multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。

但在使用这些共享API的时候，我们要注意以下几点:

在UNIX平台上，当某个进程终结之后，该进程需要被其父进程调用wait，否则进程成为僵尸进程(Zombie)。所以，有必要对每个Process对象调用join()方法 (实际上等同于wait)。对于多线程来说，由于只有一个进程，所以不存在此必要性。 multiprocessing提供了threading包中没有的IPC(比如Pipe和Queue)，效率上更高。应优先考虑Pipe和Queue，避免使用Lock/Event/Semaphore/Condition等同步方式 (因为它们占据的不是用户进程的资源)。多进程应该避免共享资源。在多线程中，我们可以比较容易地共享资源，比如使用全局变量或者传递参数。在多进程情况下，由于每个进程有自己独立的内存空间，以上方法并不合适。此时我们可以通过共享内存和Manager的方法来共享资源。但这样做提高了程序的复杂度，并因为同步的需要而降低了程序的效率。

Process.PID中保存有PID，如果进程还没有start()，则PID为None。

window系统下，需要注意的是要想启动一个子进程，必须加上那句if name == "main"，进程相关的要写在这句下面。

简单创建多进程点击查看代码1：直接在for循环中将Process类一个初始化，把需要并行化的函数作为类的输入，然后start该对象即可 from multiprocessing import Process import threading import time def foo(i): print 'say hi', i if __name__ == '__main__': for i in range(10): #创建10个进程 p = Process(target=foo, args=(i,)) p.start() 点击查看代码1运行结果 say hi 4 say hi 3 say hi 5 say hi 2 say hi 1 say hi 6 say hi 0 say hi 7 say hi 8 say hi 9 Process finished with exit code 0 #可以看出多个进程随机顺序执行点击查看代码2：将Procee生成一个自定义的派生类，在派生类中自定义run函数 from multiprocessing import Process import time class MyProcess(Process): def __init__(self, arg): super(MyProcess, self).__init__() self.arg = arg def run(self): print 'say hi', self.arg time.sleep(1) if __name__ == '__main__': for i in range(10): p = MyProcess(i) p.start() 三、 jobLib.Parallel函数

Joblib:将Python代码转换为并行计算模式，可以大大简化我们写并行计算代码的步骤.过操作该包内的函数来实现目标代码的并行计算，从而提高代码运行效率。

3.1 例子 3.1.1 不并行操作

首先，定义一个简单的函数single(a)，该函数顺序执行休眠1s然后打印a的值的操作：

from joblib import Parallel, delayed import time def single(a): """ 定义一个简单的函数 """ time.sleep(1) # 休眠1s print(a) # 打印出a

然后使用for循环运行10次single()函数，并记录运行的时间，由结果可知，这种情况下代码大概会运行10s。

start = time.time() # 记录开始的时间 for i in range(10): # 执行10次single()函数 single(i) Time = time.time() - start # 计算执行的时间 print(str(Time)+'s') # 运行结果如下 # 0 1 2 3 4 5 6 7 8 9 10.0172278881073s

不并行操作的时候，一个函数操作是1s，则运行多少次就得花多少倍的时间。

3.1.2 使用Parallel包来并行操作 Parallel函数会创建一个进程池，以便在多进程中执行每一个列表项。函数中，我们设置参数n_jobs来设置开启进程数。函数delayed是一个创建元组(function, args, kwargs)的简单技巧，比如下面代码中的意思是创建10个实参分别为0~9的single()函数的workers。 start = time.time() # 记录开始的时间 Parallel(n_jobs=3)(delayed(single)(i) for i in range(10)) # 并行化处理 Time = time.time() - start # 计算执行的时间 print(str(Time)+'s') # 运行结果如下 # 0 1 2 3 4 5 6 7 8 9 4.833665370941162s

可见并行化处理后，运行时间相比顺序执行大大减小。由于进程切换等操作的时间开销，最终的执行时间并不是理想的3.33s，而是大于一个3.33s的时间。当n_jobs的值为1时，即相当于for循环的顺序执行，结果仍然会是10s。因此我们可以改变不同的n_jobs值来查看最终的运行结果。

3.2 Parallel函数介绍 3.2.1 Parallel函数的定义方式： class joblib.parallel(n_jobs=None, backend=None, verbose=0, timeout=None, pre_dispatch='2 * n_jobs', batch_size='auto',temp_folder=None, max_nbytes='1M', mmap_mode='r', prefer=None, require=None)

Parallel参数众多，但常用的基本只有n_jobs和backend参数。

3.2.2 n_jobs: int, default: None —— 设置并行执行任务的最大数量。

当backend="multiprocessing"时指python工作进程的数量，或者backend="threading"时指线程池大小。当n_jobs=-1时，使用所有的CPU执行并行计算。当n_jobs=1时，就不会使用并行代码，即等同于顺序执行，可以在debug情况下使用。另外，当n_jobs

【本文地址】

公司简介

联系我们