python多进程程序设计 之二

ProcessPoolExecutor

ProcessPoolExecutor 类是 Executor 子类，它使用进程池异步执行调用。 ProcessPoolExecutor 使用multiprocessing模块，这允许它绕过全局解释器锁，但也意味着只能执行和返回可序列化的对象。

main 模块必须可由工作子进程导入。这意味着 ProcessPoolExecutor 将无法在交互式解释器中工作。

从提交给 ProcessPoolExecutor 的可调用对象中调用 Executor 或 Future 方法将导致死锁。

构造器

词法： concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=(), max_tasks_per_child=None)

ProcessPoolExecutor类是Executor子类，使用最多 max_workers 进程池异步执行调用。

• 如果max_workers为None，或未给出，则它将默认为计算机上的处理器数量。如果 max_workers 小于或等于 0，则会产生ValueError。
  在 Windows 上，max_workers 必须小于或等于 61。如果不是，则将产生ValueError。如果max_workers为None，则默认选择的数量最多为 61，即使有更多处理器可用。
• mp_context 可以是multiprocessing上下文，或None。它将用于启动工作进程。如果 mp_context为None，或未给出，则使用默认的multiprocessing上下文。
• initializer， initializer是一个可选的可调用对象，在每个工作进程启动时调用；
• initargs 是传递给initializer程序的参数元组。如果initializer程序产生异常，则所有当前挂起的作业，以及向池提交更多作业的任何尝试，都将产生BrokenProcessPool。
• max_tasks_per_child 是一个可选参数，指定单个进程在退出，并被新的工作进程替换之前，可以执行的最大任务数。默认情况下 max_tasks_per_child 为 None，这意味着工作进程将一直存在，直到池也消亡。

成员函数map

词法：map(fn, *iterables, timeout=None, chunksize=1)

如果调用 next()，并且在原始调用到Executor.map() 的timeout 秒数后，结果仍不可用，则返回的迭代器产生TimeoutError。 timeout 可以是 int 或 float。如果未指定timeout，或指定为None，则没有等待时间限制。

如果 fn 调用引发异常，则迭代器取值时，产生异常。

使用 ProcessPoolExecutor 时，此方法将可迭代对象分割成多个块，并将这些块作为单独的任务提交到池中。这些块的（近似）大小可以通过将 chunksize 设置为正整数来指定。对于非常长的可迭代对象，与默认大小 1 相比，使用较大的 chunksize 值，可以显着提高性能。

import time
import concurrent.futures
from multiprocessing import *

val_set = [1, 2, 3, 4, 5, 6, 7]
 
def cube(val):
    result = val * val * val
    time.sleep(0.5)
    print("[Process ID]:{0} cube({1}) = {2}".format(current_process().name, val, result))
    return [val, result]

def main(): 
    result = []
    with concurrent.futures.ProcessPoolExecutor(3) as exe:
        ret_val = exe.map(cube, val_set)
   
        for v in ret_val:
            result.append(v)

    print(result)

if __name__ == "__main__":
    freeze_support()
    main()

[Process ID]:SpawnProcess-1 cube(1) = 1
[Process ID]:SpawnProcess-2 cube(2) = 8
[Process ID]:SpawnProcess-3 cube(3) = 27
[Process ID]:SpawnProcess-1 cube(4) = 64
[Process ID]:SpawnProcess-2 cube(5) = 125
[Process ID]:SpawnProcess-3 cube(6) = 216
[Process ID]:SpawnProcess-1 cube(7) = 343
[[1, 1], [2, 8], [3, 27], [4, 64], [5, 125], [6, 216], [7, 343]]

成员函数submit

词法：submit(fn, *args, **kwargs)

规划可调用函数fn运行，*args, **kwargs作为函数的产生， 即，fn(*args, **kwargs) ，并返回一个Future 对象，该Future 对象表示可调用函数执行情况 。

from concurrent.futures import *
from multiprocessing import *

def cube(v1, v2, width, length):
    print("v1:{0}  v2:{3}  width:{1} length:{2}".format(v1, width, length, v2))
    return [v1*v2, width * length]

def main():    
    with ProcessPoolExecutor(max_workers=2) as executor:
        args = [3, 5]
        kwargs = {"width":10, "length":30}
        future = executor.submit(cube, *args, **kwargs)
        print(future.result())
    
if __name__ == "__main__":
    freeze_support()
    main()

v1:3  v2:5  width:10 length:30
[15, 300]

实列代码

import math
from multiprocessing import *
from concurrent.futures import *

PRIMES = [
    112272635095293,
    112582705942171,
    112217253509529,
    115280095190773,
    115797848077099,
    
    215797848077091,
    315797848077095,
    415797848077097,
    
    199726899285419]

def is_prime(n):
    print("Process: {0} n: {1}".format(current_process().name, n))
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    result=[]
    
    with ProcessPoolExecutor(max_workers=3) as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            result.append([number, prime])

    for number, prime in result:
        print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()

Process: SpawnProcess-1 n: 112272635095293
Process: SpawnProcess-1 n: 112582705942171
Process: SpawnProcess-2 n: 112217253509529
Process: SpawnProcess-2 n: 115280095190773
Process: SpawnProcess-3 n: 115797848077099
Process: SpawnProcess-1 n: 215797848077091
Process: SpawnProcess-1 n: 315797848077095
Process: SpawnProcess-1 n: 415797848077097
Process: SpawnProcess-1 n: 199726899285419
112272635095293 is prime: False
112582705942171 is prime: True
112217253509529 is prime: False
115280095190773 is prime: True
115797848077099 is prime: True
215797848077091 is prime: False
315797848077095 is prime: False
415797848077097 is prime: False
199726899285419 is prime: False