Python 并发模块
multiprocessing 介绍
multiprocessing 是一个用与 threading 模块相似API的支持产生进程的包。 multiprocessing 包同时提供本地和远程并发，使用子进程代替线程，有效避免 Global Interpreter Lock 带来的影响。因此， multiprocessing 模块允许程序员充分利用机器上的多个核心。Unix 和 Windows 上都可以运行。
multiprocessing 模块还引入了在 threading 模块中没有类似物的API。这方面的一个主要例子是 Pool 对象，它提供了一种方便的方法，可以跨多个输入值并行化函数的执行，跨进程分配输入数据（数据并行）。以下示例演示了在模块中定义此类函数的常见做法，以便子进程可以成功导入该模块。这个数据并行的基本例子使用 Pool
from multiprocessing import Pool
def f(x):
return x*x
if __name__ == '__main__':
with Pool(5) as p:
print(p.map(f, [1, 2, 3]))
输出结果：
[1, 4, 9]
Process 类
在 multiprocessing 中，通过创建一个 Process 对象然后调用它的 start() 方法来生成进程。 Process 和 threading.Thread API 相同。 一个简单的多进程程序示例是:from multiprocessing import Process
def f(name):
print('hello', name)
if __name__ == '__main__':
p = Process(target=f, args=('bob',))
p.start()
p.join()
输出结果：
hello bob要显示所涉及的各个进程ID，这是一个扩展示例:
from multiprocessing import Process
import os
def info(title):
print(title)
print('module name:', __name__)
print('parent process:', os.getppid())
print('process id:', os.getpid())
def f(name):
info('function f')
print('hello', name)
if __name__ == '__main__':
info('main line')
p = Process(target=f, args=('bob',))
p.start()
p.join()
输出结果：
main line module name: __main__ parent process: 817 process id: 92734 function f module name: __main__ parent process: 92734 process id: 92806 hello bob为了解释为什么
if __name__ == '__main__' 部分是必需的，请参见 编程指南。
上下文和启动方法
根据不同的平台， multiprocessing 支持三种启动进程的方法。这些 启动方法 有- spawn 父进程启动一个新的Python解释器进程。子进程只会继承那些运行进程对象的 run() 方法所需的资源。特别是父进程中非必须的文件描述符和句柄不会被继承。相对于使用 fork 或者 forkserver，使用这个方法启动进程相当慢。 允许在 Unix and Windows. 默认Windows.
- fork 父进程使用 os.fork() 来产生 Python 解释器分叉。子进程在开始时实际上与父进程相同。父进程的所有资源都由子进程继承。请注意，安全分叉多线程进程是棘手的。 只存在于Unix。Unix中的默认值。
- forkserver 程序启动并选择* forkserver * 启动方法时，将启动服务器进程。从那时起，每当需要一个新进程时，父进程就会连接到服务器并请求它分叉一个新进程。分叉服务器进程是单线程的，因此使用 os.fork() 是安全的。没有不必要的资源被继承。可在Unix平台上使用，支持通过Unix管道传递文件描述符。
if __name__ == '__main__' 子句中调用 set_start_method()。例如：
import multiprocessing as mp
def foo(q):
q.put('hello')
if __name__ == '__main__':
mp.set_start_method('spawn')
q = mp.Queue()
p = mp.Process(target=foo, args=(q,))
p.start()
print(q.get())
p.join()
输出报错：
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
File "/usr/local/Cellar/python/3.7.4/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/context.py", line 242, in set_start_method
raise RuntimeError('context has already been set')
RuntimeError: context has already been set
在程序中 set_start_method() 不应该被多次调用。
或者，你可以使用 get_context() 来获取上下文对象。上下文对象与多处理模块具有相同的API，并允许在同一程序中使用多个启动方法。:
import multiprocessing as mp
def foo(q):
q.put('hello')
if __name__ == '__main__':
ctx = mp.get_context('spawn')
q = ctx.Queue()
p = ctx.Process(target=foo, args=(q,))
p.start()
print(q.get())
p.join()
输出报错：
Traceback (most recent call last):
File "<string>", line 1, in <module>
File "/usr/local/Cellar/python/3.7.4/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/spawn.py", line 105, in spawn_main
exitcode = _main(fd)
File "/usr/local/Cellar/python/3.7.4/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/spawn.py", line 115, in _main
self = reduction.pickle.load(from_parent)
AttributeError: Can't get attribute 'foo' on <module '__main__' (built-in)>
请注意，与一个上下文相关的对象可能与不同上下文的进程不兼容。特别是，使用 fork 上下文创建的锁不能传递给使用 spawn 或 forkserver 启动方法启动的进程。
想要使用特定启动方法的库应该使用 get_context() 以避免干扰库用户的选择。
警告： 'spawn' 和 'forkserver' 启动方法当前不能在Unix上和“冻结的”可执行内容一同使用（例如，有类似 PyInstaller 和 cx_Freeze 的包产生的二进制文件）。 'fork' 启动方法可以使用。
在进程之间交换对象
multiprocessing 支持进程之间的两种通信通道： 队列 Queue 类是一个近似 queue.Queue 的克隆。 例如:from multiprocessing import Process, Queue
def f(q):
q.put([42, None, 'hello'])
if __name__ == '__main__':
q = Queue()
p = Process(target=f, args=(q,))
p.start()
print(q.get()) # prints "[42, None, 'hello']"
p.join()
队列是线程和进程安全的。
管道
Pipe() 函数返回一个由管道连接的连接对象，默认情况下是双工（双向）。例如:
from multiprocessing import Process, Pipe
def f(conn):
conn.send([42, None, 'hello'])
conn.close()
if __name__ == '__main__':
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(child_conn,))
p.start()
print(parent_conn.recv()) # prints "[42, None, 'hello']"
p.join()
返回的两个连接对象 Pipe() 表示管道的两端。每个连接对象都有 send() 和 recv() 方法（相互之间的）。请注意，如果两个进程（或线程）同时尝试读取或写入管道的 同一 端，则管道中的数据可能会损坏。当然，同时使用管道的不同端的进程不存在损坏的风险。
进程之间的同步
multiprocessing 包含来自 threading 的所有同步。例如，可以使用锁来确保一次只有一个进程打印到标准输出:from multiprocessing import Process, Lock
def f(l, i):
l.acquire()
try:
print('hello world', i)
finally:
l.release()
if __name__ == '__main__':
lock = Lock()
for num in range(10):
Process(target=f, args=(lock, num)).start()
输出结果：
hello world 0 hello world 1 hello world 3 hello world 2 hello world 4 hello world 5 hello world 6 hello world 7 hello world 8 hello world 9不使用来自不同进程的锁输出容易产生混淆。
在进程之间共享状态
如上所述，在进行并发编程时，通常最好尽量避免使用共享状态。使用多个进程时尤其如此。 但是，如果你真的需要使用一些共享数据，那么 multiprocessing 提供了两种方法。 共享内存可以使用 Value 或 Array 将数据存储在共享内存映射中。例如，以下代码:
from multiprocessing import Process, Value, Array
def f(n, a):
n.value = 3.1415927
for i in range(len(a)):
a[i] = -a[i]
if __name__ == '__main__':
num = Value('d', 0.0)
arr = Array('i', range(10))
p = Process(target=f, args=(num, arr))
p.start()
p.join()
print(num.value)
print(arr[:])
打印结果：
3.1415927 [0, -1, -2, -3, -4, -5, -6, -7, -8, -9]创建 num 和 arr 时使用的 'd' 和 'i' 参数是 array 模块使用的类型的 typecode ： 'd' 表示双精度浮点数， 'i' 表示有符号整数。这些共享对象将是进程和线程安全的。 为了更灵活地使用共享内存，可以使用 multiprocessing.sharedctypes 模块，该模块支持创建从共享内存分配的任意ctypes对象。 服务器进程
由 Manager() 返回的管理器对象控制一个服务器进程，该进程保存Python对象并允许其他进程使用代理操作它们。
Manager() 返回的管理器支持类型： list 、 dict 、 Namespace 、 Lock 、 RLock 、 Semaphore 、 BoundedSemaphore 、 Condition 、 Event 、 Barrier 、 Queue 、 Value 和 Array 。例如
from multiprocessing import Process, Manager
def f(d, l):
d[1] = '1'
d['2'] = 2
d[0.25] = None
l.reverse()
if __name__ == '__main__':
with Manager() as manager:
d = manager.dict()
l = manager.list(range(10))
p = Process(target=f, args=(d, l))
p.start()
p.join()
print(d)
print(l)
输出结果：
{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]
服务器进程管理器比使用共享内存对象更灵活，因为它们可以支持任意对象类型。此外，单个管理器可以通过网络由不同计算机上的进程共享。但是，它们比使用共享内存慢。
进程池使用
Pool 类表示一个工作进程池。它具有允许以几种不同方式将任务分配到工作进程的方法。 例如：from multiprocessing import Pool, TimeoutError
import time
import os
def f(x):
return x*x
if __name__ == '__main__':
# start 4 worker processes
with Pool(processes=4) as pool:
# print "[0, 1, 4,..., 81]"
print(pool.map(f, range(10)))
# print same numbers in arbitrary order
for i in pool.imap_unordered(f, range(10)):
print(i)
# evaluate "f(20)" asynchronously
res = pool.apply_async(f, (20,)) # runs in *only* one process
print(res.get(timeout=1)) # prints "400"
# evaluate "os.getpid()" asynchronously
res = pool.apply_async(os.getpid, ()) # runs in *only* one process
print(res.get(timeout=1)) # prints the PID of that process
# launching multiple evaluations asynchronously *may* use more processes
multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
print([res.get(timeout=1) for res in multiple_results])
# make a single worker sleep for 10 secs
res = pool.apply_async(time.sleep, (10,))
try:
print(res.get(timeout=1))
except TimeoutError:
print("We lacked patience and got a multiprocessing.TimeoutError")
print("For the moment, the pool remains available for more work")
# exiting the 'with'-block has stopped the pool
print("Now the pool is closed and no longer available")
输出结果：
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81] 0 4 16 9 1 25 36 64 81 49 400 95759 [95760, 95761, 95758, 95759] We lacked patience and got a multiprocessing.TimeoutError For the moment, the pool remains available for more work Now the pool is closed and no longer available
请注意：Pool 的方法只能由创建它的进程使用。
注解: 该软件包中的功能要求子项可以导入 __main__ 模块。这包含在 Programming guidelines 中，但值得指出。这意味着一些示例，例如 multiprocessing.pool.Pool 示例在交互式解释器中不起作用。例如:参考：https://docs.python.org/2/library/multiprocessing.html>>> from multiprocessing import Pool >>> p = Pool(5) >>> def f(x): ... return x*x ... >>> p.map(f, [1,2,3]) Process PoolWorker-1: Process PoolWorker-2: Process PoolWorker-3: Traceback (most recent call last): AttributeError: 'module' object has no attribute 'f' AttributeError: 'module' object has no attribute 'f' AttributeError: 'module' object has no attribute 'f'（如果你尝试这个，它实际上会以半随机的方式输出三个完整的回溯，然后你可能不得不以某种方式停止主进程。）
