引言

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(funciterable[, chunksize])方法将iterable对象分成一些块,作为单独的任务提交给进程池。 这些块的(近似)大小可以通过将chunksize设置为正整数来指定, 并且该方法是阻塞的。如果可迭代对象很多时,会消耗较大的内存,可以考虑使用imap或imap_unordered。

  • map_async(funciterable[, chunksize[, callback[, error_callback]]])方法是map的变种,是非阻塞的。

  • imap(funciterable[, chunksize])该方法和map一样,只不过该方法适用于对大量数据的遍历,返回的结果顺序和输入相同。

  • imap_unordered(funciterable[, 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

Multiprocessing官方文档

参考博客