Python编程之多进程(multiprocessing)详解
引言
multiprocessing是一个用于产生多进程的包,与threading模块的API类似。multiprocessing既可以实现本地的多进程,也可以实现远程的多进程。通过使用多个子进程而非线程可以绕开Python的全局解释器锁(GIL),同时允许在多种系统平台使用。
1. Process 模块
1.1 Process介绍
Process模块是一个创建进程的模块,可以通过Process直接创建进程。
multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None) """ 参数介绍: 1. group默认为None(目前未使用) 2. target代表调用对象,即子进程执行的任务 3. name为进程名称 4. args调用对象的位置参数元组,args=(value1, value2, ...) 5. kwargs调用对象的字典,kwargs={key1:value1, key2:value2, ...}
6. daemon表示进程是否为守护进程,布尔值
方法介绍:
Process.start() 启动进程,并调用子进程中的run()方法
Process.run() 进程启动时运行的方法,在自定义时必须要实现该方法
Process.terminate() 强制终止进程,不进行清理操作,如果Process创建了子进程,会导致该进程变成僵尸进程
Process.join() 阻塞进程使主进程等待该进程终止
Process.kill() 与terminate()相同
Process.is_alive() 判断进程是否还存活,如果存活,返回True
Process.close() 关闭进程对象,并清理资源,如果进程仍在运行则返回错误
"""
注意:
- 在Windows中,由于没有fork(Linux中创建进程的机制),在创建进程的时候会import启动该文件,而在import文件的时候又会再次运行整个文件,如果把Process()放在 if __name__ == ‘__main__’ 判断之外,则Process()在被import的时候也会被运行,导致无限递归创建子进程导致报错,所以在Windows系统下,必须把Process()放在 if __name__ == ‘__main__’ 的判断保护之下。
- 在子进程中不能使用input,因为输入台只显示在主进程中,故如果在子进程中使用input,会导致报错。
1.2 Process实例
from multiprocessing import Process def main(name): print(f'{name}: Hello World') if __name__ == '__main__': # 创建子进程 p = Process(target=main, args=('LovefishO',)) # 开始进程 p.start() # 阻塞进程 p.join()
例子
1.3 Process类实现
from multiprocessing import Process class NewProcess(Process): def __init__(self, name): # 执行父类的init() super().__init__() # 创建新参数 self.name = name # 在自定义Process类时,必须实现run()方法 def run(self): print(f'{self.name}: Hello World') if __name__ == '__main__': # 创建一个新的子进程,并传入参数 np = NewProcess('LovefishO') # 开始子进程 np.start() # 加入阻塞,保证主进程在子进程之后结束 np.join() print('主进程结束') # LovefishO: Hello World # 主进程结束
1.4 守护进程
正常情况下,当子进程和主进程都结束时,程序才会结束。但是当我们需要在主进程结束时,由该主进程创建的子进程也必须跟着结束时,就需要使用守护进程。当一个子进程为守护进程时,在主进程结束时,该子进程也会跟着结束。
from multiprocessing import Process def main(name): print(f'{name}: Hello World') if __name__ == '__main__': # 创建守护进程, 设置daemon = True p = Process(target=main, daemon=True, args=('LovefishO',)) # 开始进程 p.start() # 阻塞进程 p.join()
2. Pool 模块
2.1 Pool介绍
Pool模块控制着一个进程池,池中是可以执行很多任务的进程
multiprocessing.Pool(processes=None, initializer=None, initargs=(), maxtasksperchild=None, context=None) """ 参数介绍: processes: 设置要使用的进程数量,如果 processes 为 None,则使用 os.cpu_count() 返回的值
initializer: 是每个工作进程启动时要执行的可调用对象,默认为None maxtasksperchild: 工作进程退出之前可以完成的任务数,完成后用一个新的工作进程来替代原进程,为了释放闲置资源 context: 可被用于指定启动的工作进程的上下文 """
2.2 Pool中分派任务的方式
-
apply(func[, args[, kwds]])方法是阻塞,意味着当前的进程没有执行完的话,后续的进程需要等待该进程执行结束才能执行,实际上该方法是串行。
-
apply_async(func[, args[, kwds[, callback[, error_callback]]]])方法是异步非阻塞的,意味着不用等待当前进程执行完成,即可根据系统的调度切换进程,该方法是并行。
-
map(func, iterable[, chunksize])方法将iterable对象分成一些块,作为单独的任务提交给进程池。 这些块的(近似)大小可以通过将chunksize设置为正整数来指定, 并且该方法是阻塞的。如果可迭代对象很多时,会消耗较大的内存,可以考虑使用imap或imap_unordered。
-
map_async(func, iterable[, chunksize[, callback[, error_callback]]])方法是map的变种,是非阻塞的。
-
imap(func, iterable[, chunksize])该方法和map一样,只不过该方法适用于对大量数据的遍历,返回的结果顺序和输入相同。
-
imap_unordered(func, iterable[, chunksize])与imap()一样,只不过输出的顺序是任意的
2.3 Pool实例
from multiprocessing import Pool def main(name, num): print(f'{num} {name}: Hello World') if __name__ == '__main__': # 创建进程池 p = Pool() for i in range(5): p.apply(func=main, args=('LovefishO', i, )) # 关闭进程池 p.close() # 阻塞进程, 等待子进程执行结束 p.join() print('主进程结束') # 0 LovefishO: Hello World # 1 LovefishO: Hello World # 2 LovefishO: Hello World # 3 LovefishO: Hello World # 4 LovefishO: Hello World # 主进程结束
Pool例子
3. Queue 模块
3.1 Queue介绍
由于进程彼此之间互相隔离,要实现进程间通信,multiprocessing提供了两种形式:队列(queue)和管道。队列可以简单的理解为一种特殊的列表,可以设置固定的长度,从左边插入数据,从右边获取数据,并满足先进先出。并且队列时进程安全的,即同一时刻只有一个进程能够对队列进行操作。
multiprocessing.Queue(maxsize) """ 参数介绍: maxsize:设置队列长度 方法介绍: qsize():返回队列长度,该结果是不可靠的,因为在使用该方法时,队列中的数据仍在进行删除增加 empty():如果队列为空,则返回True,反之亦然,该状态不可靠 full():如果队列满了,则返回True,反之亦然,该状态不可靠 put(obj[, block[, timeout]]):将obj放入队列 get([block[, timeout]]):从队列中取出并返回对象
close(): 当前进程不会在队列中放入对象 """
3.2 Queue实例
from multiprocessing import Process, Queue def product(queue, num): # 把obj插入队列 queue.put(num) print(f'Product {num}') def consumer(queue): # 从队列中获取数据 temp = queue.get() print(f'consumer consumed {temp} product') if __name__ == '__main__': # 创建队列 q = Queue() # 生产商品 for i in range(5): p1 = Process(target=product, args=(q, i, )) p1.start() # 消费生产的商品 for i in range(5): p2 = Process(target=consumer, args=(q,)) p2.start()
Queue例子
注意:
- 在使用队列时(Queue)如果要使用进程池则不能使用multiprocessing.Pool,而是应用使用multiprocessing.Manager().Pool()
3.3 Queue + Pool实例
import multiprocessing def product(queue, num): # 把obj插入队列 queue.put(num) print(f'Product {num}') def consumer(queue): # 从队列中获取obj num = queue.get() print(f'consumer consumed {num} product') if __name__ == '__main__': # 创建队列 q = multiprocessing.Manager().Queue() # 创建进程池 p = multiprocessing.Pool() # 生产商品 for i in range(5): p.apply(func=product, args=(q, i,)) # 消费生产的商品 for i in range(5): p.apply(func=consumer, args=(q,)) # 关闭进程池 p.close() # 阻塞进程 p.join() print('主进程结束') # Product 0 # Product 1 # Product 2 # Product 3 # Product 4 # consumer consumed 0 product # consumer consumed 1 product # consumer consumed 2 product # consumer consumed 3 product # consumer consumed 4 product # 主进程结束
Queue + Pool例子
4. Reference
本文来自博客园,作者:LoveFishO,转载请注明原文链接:https://www.cnblogs.com/lovefisho/p/16202006.html