17.2. multiprocessing — 基于进程的并行¶
17.2.1. 概述¶
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]
17.2.1.1. 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()
要显示所涉及的各个进程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()
为了解释为什么 if __name__ == '__main__' 部分是必需的，请参见 编程指导。
17.2.1.2. 上下文和启动方法¶
根据不同的平台， multiprocessing 支持三种启动进程的方法。这些 启动方法 有
- spawn
父进程启动一个新的Python解释器进程。子进程只会继承那些运行进程对象的
run()方法所需的资源。特别是父进程中非必须的文件描述符和句柄不会被继承。相对于使用 fork 或者 forkserver，使用这个方法启动进程相当慢。可在Unix和Windows上使用。 Windows上的默认设置。
- fork
父进程使用
os.fork()来产生 Python 解释器分叉。子进程在开始时实际上与父进程相同。父进程的所有资源都由子进程继承。请注意，安全分叉多线程进程是棘手的。只存在于Unix。Unix中的默认值。
- forkserver
程序启动并选择* forkserver * 启动方法时，将启动服务器进程。从那时起，每当需要一个新进程时，父进程就会连接到服务器并请求它分叉一个新进程。分叉服务器进程是单线程的，因此使用
os.fork()是安全的。没有不必要的资源被继承。可在Unix平台上使用，支持通过Unix管道传递文件描述符。
在 3.4 版更改: spawn 在所有unix平台上添加，并且为一些unix平台添加了 forkserver 。子进程不再继承Windows上的所有上级进程可继承的句柄。
On Unix using the spawn or forkserver start methods will also start a semaphore tracker process which tracks the unlinked named semaphores created by processes of the program. When all processes have exited the semaphore tracker unlinks any remaining semaphores. Usually there should be none, but if a process was killed by a signal there may some “leaked” semaphores. (Unlinking the named semaphores is a serious matter since the system allows only a limited number, and they will not be automatically unlinked until the next reboot.)
要选择一个启动方法，你应该在主模块的 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()
在程序中 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()
Note that objects related to one context may not be compatible with processes for a different context. In particular, locks created using the fork context cannot be passed to a processes started using the spawn or forkserver start methods.
想要使用特定启动方法的库应该使用 get_context() 以避免干扰库用户的选择。
17.2.1.3. 在进程之间交换对象¶
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()方法（相互之间的）。请注意，如果两个进程（或线程）同时尝试读取或写入管道的 同一 端，则管道中的数据可能会损坏。当然，同时使用管道的不同端的进程不存在损坏的风险。
17.2.1.4. 进程之间的同步¶
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()
不使用来自不同进程的锁输出容易产生混淆。
17.2.1.5. 在进程之间共享状态¶
如上所述，在进行并发编程时，通常最好尽量避免使用共享状态。使用多个进程时尤其如此。
但是，如果你真的需要使用一些共享数据，那么 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]服务器进程管理器比使用共享内存对象更灵活，因为它们可以支持任意对象类型。此外，单个管理器可以通过网络由不同计算机上的进程共享。但是，它们比使用共享内存慢。
17.2.1.6. 使用工作进程¶
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")
请注意，池的方法只能由创建它的进程使用。
注解
该软件包中的功能要求子项可以导入 __main__ 模块。这包含在 编程指导 中，但值得指出。这意味着一些示例，例如 multiprocessing.pool.Pool 示例在交互式解释器中不起作用。例如:
>>> 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'
（如果你尝试这个，它实际上会以半随机的方式输出三个完整的回溯，然后你可能不得不以某种方式停止主进程。）
17.2.2. 参考¶
multiprocessing 包大部分复制了 threading 模块的API。
17.2.2.1. Process 和异常¶
-
class
multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)¶ 进程对象表示在单独进程中运行的活动。
Process类等价于threading.Thread。应始终使用关键字参数调用构造函数。 group 应该始终是
None；它仅用于兼容threading.Thread。 target 是由run()方法调用的可调用对象。它默认为None，意味着什么都没有被调用。 name 是进程名称（有关详细信息，请参阅name）。 args 是目标调用的参数元组。 kwargs 是目标调用的关键字参数字典。如果提供，则键参数 daemon 将进程daemon标志设置为True或False。如果是None（默认值），则该标志将从创建的进程继承。默认情况下，不会将任何参数传递给 target 。
如果子类重写构造函数，它必须确保它在对进程执行任何其他操作之前调用基类构造函数（
Process.__init__()）。在 3.3 版更改: 加入 daemon 参数。
-
run()¶ 表示进程活动的方法。
你可以在子类中重载此方法。标准
run()方法调用传递给对象构造函数的可调用对象作为目标参数（如果有），分别从 args 和 kwargs 参数中获取顺序和关键字参数。
-
join([timeout])¶ 如果可选参数 timeout 是
None（默认值），则该方法将阻塞，直到调用join()方法的进程终止。如果 timeout 是一个正数，它最多会阻塞 timeout 秒。请注意，如果进程终止或方法超时，则该方法返回None。检查进程的exitcode以确定它是否终止。一个进程可以合并多次。
进程无法并入自身，因为这会导致死锁。尝试在启动进程之前合并进程是错误的。
-
name¶ 进程的名称。该名称是一个字符串，仅用于识别目的。它没有语义。可以为多个进程指定相同的名称。
初始名称由构造器设定。 如果没有为构造器提供显式名称，则会构造一个形式为 ‘Process-N1:N2:…:Nk’ 的名称，其中每个 Nk 是其父亲的第 N 个孩子。
-
daemon¶ 进程的守护标志，一个布尔值。这必须在
start()被调用之前设置。初始值继承自创建进程。
当进程退出时，它会尝试终止其所有守护进程子进程。
请注意，不允许守护进程创建子进程。否则，守护进程会在子进程退出时终止其子进程。 另外，这些 不是 Unix守护进程或服务，它们是正常进程，如果非守护进程已经退出，它们将被终止（并且不被合并）。
除了
threading.ThreadAPI ，Process对象还支持以下属性和方法：-
pid¶ 返回进程ID。在生成该进程之前，这将是
None。
-
exitcode¶ 的退子进程出代码。如果进程尚未终止，这将是
None。负值 -N 表示孩子被信号 N 终止。
-
authkey¶ 进程的身份验证密钥（字节字符串）。
当
multiprocessing初始化时，主进程使用os.urandom()分配一个随机字符串。当创建
Process对象时，它将继承其父进程的身份验证密钥，尽管可以通过将authkey设置为另一个字节字符串来更改。参见 认证密码 。
-
sentinel¶ 系统对象的数字句柄，当进程结束时将变为 “ready” 。
如果要使用
multiprocessing.connection.wait()一次等待多个事件，可以使用此值。否则调用join()更简单。在Windows上，这是一个操作系统句柄，可以与
WaitForSingleObject和WaitForMultipleObjects系列API调用一起使用。在Unix上，这是一个文件描述符，可以使用来自select模块的原语。3.3 新版功能.
-
terminate()¶ 终止进程。 在Unix上，这是使用
SIGTERM信号完成的；在Windows上使用TerminateProcess()。 请注意，不会执行退出处理程序和finally子句等。请注意，进程的后代进程将不会被终止 —— 它们将简单地变成孤立的。
警告
如果在关联进程使用管道或队列时使用此方法，则管道或队列可能会损坏，并可能无法被其他进程使用。类似地，如果进程已获得锁或信号量等，则终止它可能导致其他进程死锁。
注意
start()、join()、is_alive()、terminate()和exitcode方法只能由创建进程对象的进程调用。Process一些方法的示例用法：>>> import multiprocessing, time, signal >>> p = multiprocessing.Process(target=time.sleep, args=(1000,)) >>> print(p, p.is_alive()) <Process(Process-1, initial)> False >>> p.start() >>> print(p, p.is_alive()) <Process(Process-1, started)> True >>> p.terminate() >>> time.sleep(0.1) >>> print(p, p.is_alive()) <Process(Process-1, stopped[SIGTERM])> False >>> p.exitcode == -signal.SIGTERM True
-
-
exception
multiprocessing.ProcessError¶ 所有
multiprocessing异常的基类。
-
exception
multiprocessing.BufferTooShort¶ 当提供的缓冲区对象太小而无法读取消息时，
Connection.recv_bytes_into()引发的异常。如果
e是一个BufferTooShort实例，那么e.args[0]将把消息作为字节字符串给出。
-
exception
multiprocessing.AuthenticationError¶ 出现身份验证错误时引发。
-
exception
multiprocessing.TimeoutError¶ 有超时的方法超时时引发。
17.2.2.2. 管道和队列¶
使用多进程时，一般使用消息机制实现进程间通信，尽可能避免使用同步原语，例如锁。
消息机制包含： Pipe() (可以用于在两个进程间传递消息)，以及队列(能够在多个生产者和消费者之间通信)。
The Queue, SimpleQueue and JoinableQueue types are multi-producer,
multi-consumer FIFO queues modelled on the queue.Queue class in the
standard library. They differ in that Queue lacks the
task_done() and join() methods introduced
into Python 2.5’s queue.Queue class.
如果你使用了 JoinableQueue ，那么你**必须**对每个已经移出队列的任务调用 JoinableQueue.task_done() 。不然的话用于统计未完成任务的信号量最终会溢出并抛出异常。
另外还可以通过使用一个管理器对象创建一个共享队列，详见 数据管理器 。
注解
multiprocessing 使用了普通的 queue.Empty 和 queue.Full 异常去表示超时。 你需要从 queue 中导入它们，因为它们并不在 multiprocessing 的命名空间中。
注解
当一个对象被放入一个队列中时，这个对象首先会被一个后台线程用pickle序列化，并将序列化后的数据通过一个底层管道的管道传递到队列中。这种做法会有点让人惊讶，但一般不会出现什么问题。如果它们确实妨碍了你，你可以使用一个由管理器 manager 创建的队列替换它。
- 将一个对象放入一个空队列后，可能需要极小的延迟，队列的方法
empty()才会返回False。而get_nowait()可以不抛出queue.Empty直接返回。 - 如果有多个进程同时将对象放入队列，那么在队列的另一端接受到的对象可能是无序的。但是由同一个进程放入的多个对象的顺序在另一端输出时总是一样的。
警告
如果一个进程通过调用 Process.terminate() 或 os.kill() 在尝试使用 Queue 期间被终止了，那么队列中的数据很可能被破坏。 这可能导致其他进程在尝试使用该队列时遇到异常。
警告
正如刚才提到的，如果一个子进程将一些对象放进队列中 (并且它没有用 JoinableQueue.cancel_join_thread 方法)，那么这个进程在所有缓冲区的对象被刷新进管道之前，是不会终止的。
这意味着，除非你确定所有放入队列中的对象都已经被消费了，否则如果你试图等待这个进程，你可能会陷入死锁中。相似地，如果该子进程不是后台进程，那么父进程可能在试图等待所有非后台进程退出时挂起。
注意用管理器创建的队列不存在这个问题，详见 编程指导 。
该 例子 展示了如何使用队列实现进程间通信。
-
multiprocessing.Pipe([duplex])¶ Returns a pair
(conn1, conn2)ofConnectionobjects representing the ends of a pipe.如果 duplex 被置为
True(默认值)，那么该管道是双向的。如果 duplex 被置为False，那么该管道是单向的，即conn1只能用于接收消息，而conn2仅能用于发送消息。
-
class
multiprocessing.Queue([maxsize])¶ 返回一个使用一个管道和少量锁和信号量实现的共享队列实例。当一个进程将一个对象放进队列中时，一个写入线程会启动并将对象从缓冲区写入管道中。
一旦超时，将抛出标准库
queue模块中常见的异常queue.Empty和queue.Full。除了
task_done()和join()之外，Queue实现了标准库类queue.Queue中所有的方法。-
qsize()¶ 返回队列的大致长度。由于多线程或者多进程的上下文，这个数字是不可靠的。
注意，在 Unix 平台上，例如 Mac OS X ，这个方法可能会抛出
NotImplementedError异常，因为该平台没有实现sem_getvalue()。
-
empty()¶ 如果队列是空的，返回
True，反之返回False。 由于多线程或多进程的环境，该状态是不可靠的。
-
full()¶ 如果队列是满的，返回
True，反之返回False。 由于多线程或多进程的环境，该状态是不可靠的。
-
put(obj[, block[, timeout]])¶ 将 obj 放入队列。如果可选参数 block 是
True(默认值) 而且 timeout 是None(默认值), 将会阻塞当前进程，直到有空的缓冲槽。如果 timeout 是正数，将会在阻塞了最多 timeout 秒之后还是没有可用的缓冲槽时抛出queue.Full异常。反之 (block 是False时)，仅当有可用缓冲槽时才放入对象，否则抛出queue.Full异常 (在这种情形下 timeout 参数会被忽略)。
-
put_nowait(obj)¶ 相当于
put(obj, False)。
-
get([block[, timeout]])¶ 从队列中取出并返回对象。如果可选参数 block 是
True(默认值) 而且 timeout 是None(默认值), 将会阻塞当前进程，直到队列中出现可用的对象。如果 timeout 是正数，将会在阻塞了最多 timeout 秒之后还是没有可用的对象时抛出queue.Empty异常。反之 (block 是False时)，仅当有可用对象能够取出时返回，否则抛出queue.Empty异常 (在这种情形下 timeout 参数会被忽略)。
-
get_nowait()¶ 相当于
get(False)。
multiprocessing.Queue类有一些在queue.Queue类中没有出现的方法。这些方法在大多数情形下并不是必须的。-
close()¶ 指示当前进程将不会再往队列中放入对象。一旦所有缓冲区中的数据被写入管道之后，后台的线程会退出。这个方法在队列被gc回收时会自动调用。
-
join_thread()¶ 等待后台线程。这个方法仅在调用了
close()方法之后可用。这会阻塞当前进程，直到后台线程退出，确保所有缓冲区中的数据都被写入管道中。默认情况下，如果一个不是队列创建者的进程试图退出，它会尝试等待这个队列的后台线程。这个进程可以使用
cancel_join_thread()让join_thread()方法什么都不做直接跳过。
-
cancel_join_thread()¶ 防止
join_thread()方法阻塞当前进程。具体而言，这防止进程退出时自动等待后台线程退出。详见join_thread()。可能这个方法称为”
allow_exit_without_flush()“ 会更好。这有可能会导致正在排队进入队列的数据丢失，大多数情况下你不需要用到这个方法，仅当你不关心底层管道中可能丢失的数据，只是希望进程能够马上退出时使用。
注解
该类的功能依赖于宿主操作系统具有可用的共享信号量实现。否则该类将被禁用，任何试图实例化一个
Queue对象的操作都会抛出ImportError异常，更多信息详见 bpo-3770 。后续说明的任何专用队列对象亦如此。-
-
class
multiprocessing.SimpleQueue¶ 这是一个简化的
Queue类的实现，很像带锁的Pipe。-
empty()¶ 如果队列为空返回
True，否则返回False。
-
get()¶ 从队列中移出并返回一个对象。
-
put(item)¶ 将 item 放入队列。
-
-
class
multiprocessing.JoinableQueue([maxsize])¶ JoinableQueue类是Queue的子类，额外添加了task_done()和join()方法。-
task_done()¶ 指出之前进入队列的任务已经完成。由队列的消费者进程使用。对于每次调用
get()获取的任务，执行完成后调用task_done()告诉队列该任务已经处理完成。如果
join()方法正在阻塞之中，该方法会在所有对象都被处理完的时候返回 (即对之前使用put()放进队列中的所有对象都已经返回了对应的task_done()) 。如果被调用的次数多于放入队列中的项目数量，将引发
ValueError异常 。
-
join()¶ 阻塞至队列中所有的元素都被接收和处理完毕。
当条目添加到队列的时候，未完成任务的计数就会增加。每当消费者进程调用
task_done()表示这个条目已经被回收，该条目所有工作已经完成，未完成计数就会减少。当未完成计数降到零的时候，join()阻塞被解除。
-
17.2.2.3. 杂项¶
-
multiprocessing.active_children()¶ 返回当前进程存活的子进程的列表。
调用该方法有“等待”已经结束的进程的副作用。
-
multiprocessing.cpu_count()¶ Return the number of CPUs in the system. May raise
NotImplementedError.
-
multiprocessing.freeze_support()¶ 为使用了
multiprocessing的程序，提供冻结以产生 Windows 可执行文件的支持。(在 py2exe, PyInstaller 和 cx_Freeze 上测试通过)需要在 main 模块的
if __name__ == '__main__'该行之后马上调用该函数。例如：from multiprocessing import Process, freeze_support def f(): print('hello world!') if __name__ == '__main__': freeze_support() Process(target=f).start()
如果没有调用
freeze_support()在尝试运行被冻结的可执行文件时会抛出RuntimeError异常。对
freeze_support()的调用在非 Windows 平台上是无效的。如果该模块在 Windows 平台的 Python 解释器中正常运行 (该程序没有被冻结)， 调用``freeze_support()`` 也是无效的。
-
multiprocessing.get_all_start_methods()¶ 返回支持的启动方法的列表，该列表的首项即为默认选项。可能的启动方法有
'fork','spawn'和``’forkserver’。在 Windows 中，只有 ``'spawn'是可用的。Unix平台总是支持``’fork’`` 和``’spawn’，且 ``'fork'是默认值。3.4 新版功能.
-
multiprocessing.get_context(method=None)¶ 返回一个 Context 对象。该对象具有和
multiprocessing模块相同的API。如果 method 设置成
None那么将返回默认上下文对象。否则 method 应该是'fork','spawn','forkserver'。 如果指定的启动方法不存在，将抛出ValueError异常。3.4 新版功能.
-
multiprocessing.get_start_method(allow_none=False)¶ 返回启动进程时使用的启动方法名。
如果启动方法已经固定，并且 allow_none 被设置成 False ，那么启动方法将被固定为默认的启动方法，并且返回其方法名。如果启动方法没有设定，并且 allow_none 被设置成 True ，那么将返回
None。返回值将为
'fork','spawn','forkserver'或者None。'fork'``是 Unix 的默认值， ``'spawn'是 Windows 的默认值。3.4 新版功能.
-
multiprocessing.set_executable()¶ 设置在启动子进程时使用的 Python 解释器路径。 ( 默认使用
sys.executable) 嵌入式编程人员可能需要这样做：set_executable(os.path.join(sys.exec_prefix, 'pythonw.exe'))
以使他们可以创建子进程。
在 3.4 版更改: 现在在 Unix 平台上使用
'spawn'启动方法时支持调用该方法。
-
multiprocessing.set_start_method(method)¶ 设置启动子进程的方法。 method 可以是
'fork','spawn'或者'forkserver'。注意这最多只能调用一次，并且需要藏在 main 模块中，由
if __name__ == '__main__'保护着。3.4 新版功能.
注解
multiprocessing 并没有包含类似 threading.active_count() , threading.enumerate() , threading.settrace() , threading.setprofile(), threading.Timer , 或者 threading.local 的方法和类。
17.2.2.4. 连接对象（Connection）¶
Connection 对象允许收发可以序列化的对象或字符串。它们可以看作面向消息的连接套接字。
Connection objects are usually created using Pipe() – see also
监听者及客户端.
-
class
multiprocessing.Connection¶ -
send(obj)¶ 将一个对象发送到连接的另一端，可以用
recv()读取。The object must be picklable. Very large pickles (approximately 32 MB+, though it depends on the OS) may raise a
ValueErrorexception.
-
fileno()¶ 返回由连接对象使用的描述符或者句柄。
-
close()¶ 关闭连接对象。
当连接对象被垃圾回收时会自动调用。
-
poll([timeout])¶ 返回连接对象中是否有可以读取的数据。
如果未指定 timeout ，此方法会马上返回。如果 timeout 是一个数字，则指定了最大阻塞的秒数。如果 timeout 是
None，那么将一直等待，不会超时。注意通过使用
multiprocessing.connection.wait()可以一次轮询多个连接对象。
-
send_bytes(buffer[, offset[, size]])¶ 从一个 bytes-like object （字节类对象）对象中取出字节数组并作为一条完整消息发送。
If offset is given then data is read from that position in buffer. If size is given then that many bytes will be read from buffer. Very large buffers (approximately 32 MB+, though it depends on the OS) may raise a
ValueErrorexception
-
recv_bytes([maxlength])¶ 以字符串形式返回一条从连接对象另一端发送过来的字节数据。此方法在接收到数据前将一直阻塞。 如果连接对象被对端关闭或者没有数据可读取，将抛出
EOFError异常。如果给定了 maxlength 并且消息长于 maxlength 那么将抛出
OSError并且该连接对象将不再可读。
-
recv_bytes_into(buffer[, offset])¶ 将一条完整的字节数据消息读入 buffer 中并返回消息的字节数。 此方法在接收到数据前将一直阻塞。 如果连接对象被对端关闭或者没有数据可读取，将抛出
EOFError异常。buffer must be a writable bytes-like object. If offset is given then the message will be written into the buffer from that position. Offset must be a non-negative integer less than the length of buffer (in bytes).
如果缓冲区太小，则将引发
BufferTooShort异常，并且完整的消息将会存放在异常实例e的e.args[0]中。
在 3.3 版更改: 现在连接对象自身可以通过
Connection.send()和Connection.recv()在进程之间传递。3.3 新版功能: 连接对象现已支持上下文管理协议 – 参见 see 上下文管理器类型 。
__enter__()返回连接对象，__exit__()会调用close()。-
例如:
>>> from multiprocessing import Pipe
>>> a, b = Pipe()
>>> a.send([1, 'hello', None])
>>> b.recv()
[1, 'hello', None]
>>> b.send_bytes(b'thank you')
>>> a.recv_bytes()
b'thank you'
>>> import array
>>> arr1 = array.array('i', range(5))
>>> arr2 = array.array('i', [0] * 10)
>>> a.send_bytes(arr1)
>>> count = b.recv_bytes_into(arr2)
>>> assert count == len(arr1) * arr1.itemsize
>>> arr2
array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])
警告
Connection.recv() 方法会自动解封它收到的数据，除非你能够信任发送消息的进程，否则此处可能有安全风险。
因此， 除非连接对象是由 Pipe() 产生的，否则你应该仅在使用了某种认证手段之后才使用 recv() 和 send() 方法。 参考 认证密码。
警告
如果一个进程在试图读写管道时被终止了，那么管道中的数据很可能是不完整的，因为此时可能无法确定消息的边界。
17.2.2.5. 同步原语¶
通常来说同步原语在多进程环境中并不像它们在多线程环境中那么必要。参考 threading 模块的文档。
注意可以使用管理器对象创建同步原语，参考 数据管理器 。
-
class
multiprocessing.Barrier(parties[, action[, timeout]])¶ 类似
threading.Barrier的栅栏对象。3.3 新版功能.
-
class
multiprocessing.BoundedSemaphore([value])¶ 非常类似
threading.BoundedSemaphore的有界信号量对象。一个小小的不同在于，它的
acquire方法的第一个参数名是和Lock.acquire()一样的 block 。注解
在 Mac OS X 平台上， 该对象于
Semaphore不同在于sem_getvalue()方法并没有在该平台上实现。
-
class
multiprocessing.Condition([lock])¶ 条件变量：
threading.Condition的别名。指定的 lock 参数应该是
multiprocessing模块中的Lock或者RLock对象。在 3.3 版更改: 新增了
wait_for()方法。
-
class
multiprocessing.Event¶ A clone of
threading.Event.
-
class
multiprocessing.Lock¶ 原始锁（非递归锁）对象，类似于
threading.Lock。一旦一个进程或者线程拿到了锁，后续的任何其他进程或线程的其他请求都会被阻塞直到锁被释放。任何进程或线程都可以释放锁。除非另有说明，否则multiprocessing.Lock用于进程或者线程的概念和行为都和threading.Lock一致。注意
Lock实际上是一个工厂函数。它返回由默认上下文初始化的multiprocessing.synchronize.Lock对象。Locksupports the context manager protocol and thus may be used inwithstatements.-
acquire(block=True, timeout=None)¶ 可以阻塞或非阻塞地获得锁。
如果 block 参数被设为
True( 默认值 ) ， 对该方法的调用在锁处于释放状态之前都会阻塞，然后将锁设置为锁住状态并返回True。需要注意的是第一个参数名与threading.Lock.acquire()的不同。如果 block 参数被设置成
False，方法的调用将不会阻塞。 如果锁当前处于锁住状态，将返回False； 否则将锁设置成锁住状态，并返回True。当 timeout 是一个正浮点数时，会在等待锁的过程中最多阻塞等待 timeout 秒，当 timeout 是负数时，效果和 timeout 为0时一样，当 timeout 是
None（默认值）时，等待时间是无限长。需要注意的是，对于 timeout 是负数和None的情况, 其行为与threading.Lock.acquire()是不一样的。当 block 参数为False时， timeout 并没有实际用处，会直接忽略， 当 block 参数为True时，函数会在拿到锁后返回True或者 超时没拿到锁后返回False。
-
release()¶ 释放锁，可以在任何进程、线程使用，并不限于锁的拥有者。
其行为与
threading.Lock.release()一样，只不过当尝试释放一个没有被持有的锁时，会抛出ValueError异常。
-
-
class
multiprocessing.RLock¶ 递归锁对象: 类似于
threading.RLock。递归锁必须由持有线程、进程亲自释放。如果某个进程或者线程拿到了递归锁，这个进程或者线程可以再次拿到这个锁而不需要等待。但是这个进程或者线程的拿锁操作和释放锁操作的次数必须相同。注意
RLock是一个工厂函数，调用后返回一个使用默认 context 初始化的multiprocessing.synchronize.RLock对象。RLock支持 context manager 协议，因此可在with语句内使用。-
acquire(block=True, timeout=None)¶ 可以阻塞或非阻塞地获得锁。
当 block 设置为
True时，会一直阻塞直到锁处于空闲状态（没有被任何进程、线程拥有），除非当前进程/线程已经拥有了这把锁。然后当前进程会持有这把锁（在锁没有被持有者的情况下），锁内的递归等级加一，并返回True. 注意， 这个函数第一个参数和threading.RLock.acquire()有几个不同点，包括参数名本身。当 block 参数是
False, 将不会阻塞，如果此时锁被其他进程或者线程持有，当前进程、线程获取锁操作失败，锁的递归等级也不会改变，函数返回False, 如果当前锁已经处于释放状态，则当前进程、线程则会拿到锁，并且锁内的递归等级加一，函数返回True。timeout 参数的使用方法及行为与
Lock.acquire()一样。但是要注意 timeout 的其中一些行为和threading.RLock.acquire()中实现的行为是不同的。
-
release()¶ 释放锁，使锁内的递归等级减一。如果释放后锁内的递归等级降低为0，则会重置锁的状态为释放状态（即没有被任何进程、线程持有），重置后如果有有其他进程和线程在等待这把锁，他们中的一个会获得这个锁而继续运行。如果释放后锁内的递归等级还没到达0，则这个锁仍将保持未释放状态且当前进程和线程仍然是持有者。
只有当前进程或线程是锁的持有者时，才允许调用这个方法。如果当前进程或线程不是这个锁的拥有者，或者这个锁处于已释放的状态(即没有任何拥有者)，调用这个方法会抛出
AssertionError异常。注意这里抛出的异常类型和threading.RLock.release()中实现的行为不一样。
-
-
class
multiprocessing.Semaphore([value])¶ 一种信号量对象: 类似于
threading.Semaphore.一个小小的不同在于，它的
acquire方法的第一个参数名是和Lock.acquire()一样的 block 。
注解
在 Mac OS X 上，不支持 sem_timedwait ，所以，使用调用 acquire() 时如果使用 timeout 参数，会通过循环sleep来模拟timeout的行为。
注解
假如信号 SIGINT 是来自于 Ctrl-C ，并且主线程被 BoundedSemaphore.acquire(), Lock.acquire(), RLock.acquire(), Semaphore.acquire(), Condition.acquire() 或 Condition.wait() 阻塞，则调用会立即中断同时抛出 KeyboardInterrupt 异常。
这和 threading 的行为不同，此模块中当执行对应的阻塞式调用时，SIGINT 会被忽略。
注解
这个库的某些功能依赖于宿主机系统的共享信号量，如果系统没有这个特性， multiprocessing.synchronize 会被禁用，尝试导入这个包会引发 ImportError 异常，详细信息请查看 bpo-3770 。
17.2.2.7. 数据管理器¶
管理器提供了一种创建共享数据的方法，从而可以在不同进程中共享，甚至可以通过网络跨机器共享数据。管理器维护一个用于管理 共享对象 的服务。其他进程可以通过代理访问这些共享对象。
返回一个已启动的
SyncManager管理器对象，这个对象可以用于在不同进程中共享数据。返回的管理器对象对应了一个 spawned 方式启动的子进程，并且拥有一系列方法可以用于创建共享对象、返回对应的代理。
当管理器被垃圾回收或者父进程退出时，管理器进程会立即退出。管理器类定义在 multiprocessing.managers 模块:
-
class
multiprocessing.managers.BaseManager([address[, authkey]])¶ 创建一个 BaseManager 对象。
一旦创建，应该及时调用
start()或者get_server().serve_forever()以确保管理器对象对应的管理进程已经启动。address 是管理器对象监听的地址。如果 address 是
None,则允许和任意主机的请求建立连接。authkey 是认证标识，用于检查连接服务进程的请求合法性。如果 authkey 是
None, 则会使用current_process().authkey, 否则，就使用 authkey , 需要保证它必须是 byte 类型的字符串。-
start([initializer[, initargs]])¶ 为管理器开启一个子进程，如果 initializer 不是
None, 子进程在启动时将会调用initializer(*initargs)。
-
get_server()¶ 返回一个
Server对象，它是管理器在后台控制的真实的服务。Server对象拥有serve_forever()方法。>>> from multiprocessing.managers import BaseManager >>> manager = BaseManager(address=('', 50000), authkey=b'abc') >>> server = manager.get_server() >>> server.serve_forever()
Server额外拥有一个address属性。
-
connect()¶ 将本地管理器对象连接到一个远程管理器进程:
>>> from multiprocessing.managers import BaseManager >>> m = BaseManager(address=('127.0.0.1', 5000), authkey=b'abc') >>> m.connect()
-
register(typeid[, callable[, proxytype[, exposed[, method_to_typeid[, create_method]]]]])¶ 一个 classmethod，可以将一个类型或者可调用对象注册到管理器类。
typeid 是一种 “类型标识”，用于唯一表示某种共享类型，必须是一个字符串。
callable 是一个用来为此类型标识符创建对象的可调用对象。如果一个管理器实例将使用
connect()方法连接到服务器，或者 create_method 参数为False，那么这里可留下None。proxytype 是
BaseProxy的子类，可以根据 typeid 为共享对象创建一个代理，如果是None, 则会自动创建一个代理类。exposed 是一个函数名组成的序列，用来指明只有这些方法可以使用
BaseProxy._callmethod()代理。(如果 exposed 是None, 则会在proxytype._exposed_存在的情况下转而使用它) 当暴露的方法列表没有指定的时候，共享对象的所有 “公共方法” 都会被代理。（这里的“公共方法”是指所有拥有__call__()方法并且不是以'_'开头的属性）method_to_typeid 是一个映射，用来指定那些应该返回代理对象的暴露方法所返回的类型。（如果 method_to_typeid 是
None, 则proxytype._method_to_typeid_会在存在的情况下被使用）如果方法名称不在这个映射中或者映射是None,则方法返回的对象会是一个值拷贝。create_method 指明，是否要创建一个以 typeid 命名并返回一个代理对象的函数，这个函数会被服务进程用于创建共享对象，默认为
True。
BaseManager实例也有一个只读属性。-
address¶ 管理器所用的地址。
在 3.3 版更改: 管理器对象支持上下文管理协议 - 查看 上下文管理器类型 。
__enter__()启动服务进程（如果它还没有启动）并且返回管理器对象，__exit__()会调用shutdown()。在之前的版本中，如果管理器服务进程没有启动，
__enter__()不会负责启动它。-
-
class
multiprocessing.managers.SyncManager¶ BaseManager的子类，可用于进程的同步。这个类型的对象使用multiprocessing.Manager()创建。It also supports creation of shared lists and dictionaries.
-
Barrier(parties[, action[, timeout]])¶ 创建一个共享的
threading.Barrier对象并返回它的代理。3.3 新版功能.
-
BoundedSemaphore([value])¶ 创建一个共享的
threading.BoundedSemaphore对象并返回它的代理。
-
Condition([lock])¶ 创建一个共享的
threading.Condition对象并返回它的代理。如果提供了 lock 参数，那它必须是
threading.Lock或threading.RLock的代理对象。在 3.3 版更改: 新增了
wait_for()方法。
-
Event()¶ 创建一个共享的
threading.Event对象并返回它的代理。
-
Lock()¶ 创建一个共享的
threading.Lock对象并返回它的代理。
-
Namespace()¶ 创建一个共享的
Namespace`对象并返回它的代理。
-
Queue([maxsize])¶ 创建一个共享的
queue.Queue对象并返回它的代理。
-
RLock()¶ 创建一个共享的
threading.RLock对象并返回它的代理。
-
Semaphore([value])¶ 创建一个共享的
threading.Semaphore对象并返回它的代理。
-
Array(typecode, sequence)¶ 创建一个数组并返回它的代理。
-
Value(typecode, value)¶ 创建一个具有可写
value属性的对象并返回它的代理。
-
dict()¶ -
dict(mapping) -
dict(sequence) Create a shared
dictobject and return a proxy for it.
-
list()¶ -
list(sequence) Create a shared
listobject and return a proxy for it.
注解
Modifications to mutable values or items in dict and list proxies will not be propagated through the manager, because the proxy has no way of knowing when its values or items are modified. To modify such an item, you can re-assign the modified object to the container proxy:
# create a list proxy and append a mutable object (a dictionary) lproxy = manager.list() lproxy.append({}) # now mutate the dictionary d = lproxy[0] d['a'] = 1 d['b'] = 2 # at this point, the changes to d are not yet synced, but by # reassigning the dictionary, the proxy is notified of the change lproxy[0] = d
-
-
class
multiprocessing.managers.Namespace¶ 一个可以注册到
SyncManager的类型。命名空间对象没有公共方法，但是拥有可写的属性。它的表示(repr)会显示所有属性的值。
值得一提的是，当对命名空间对象使用代理的时候，访问所有名称以
'_'开头的属性都只是代理器上的属性，而不是命名空间对象的属性。>>> manager = multiprocessing.Manager() >>> Global = manager.Namespace() >>> Global.x = 10 >>> Global.y = 'hello' >>> Global._z = 12.3 # this is an attribute of the proxy >>> print(Global) Namespace(x=10, y='hello')
17.2.2.7.1. 自定义管理器¶
要创建一个自定义的管理器，需要新建一个 BaseManager 的子类，然后使用这个管理器类上的 register() 类方法将新类型或者可调用方法注册上去。例如:
from multiprocessing.managers import BaseManager
class MathsClass:
def add(self, x, y):
return x + y
def mul(self, x, y):
return x * y
class MyManager(BaseManager):
pass
MyManager.register('Maths', MathsClass)
if __name__ == '__main__':
with MyManager() as manager:
maths = manager.Maths()
print(maths.add(4, 3)) # prints 7
print(maths.mul(7, 8)) # prints 56
17.2.2.7.2. 使用远程管理器¶
可以将管理器服务运行在一台机器上，然后使用客户端从其他机器上访问。(假设它们的防火墙允许这样的网络通信)
运行下面的代码可以启动一个服务，此付包含了一个共享队列，允许远程客户端访问:
>>> from multiprocessing.managers import BaseManager
>>> import queue
>>> queue = queue.Queue()
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue', callable=lambda:queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()
远程客户端可以通过下面的方式访问服务:
>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.put('hello')
也可以通过下面的方式:
>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.get()
'hello'
本地进程也可以访问这个队列，利用上面的客户端代码通过远程方式访问:
>>> from multiprocessing import Process, Queue
>>> from multiprocessing.managers import BaseManager
>>> class Worker(Process):
... def __init__(self, q):
... self.q = q
... super(Worker, self).__init__()
... def run(self):
... self.q.put('local hello')
...
>>> queue = Queue()
>>> w = Worker(queue)
>>> w.start()
>>> class QueueManager(BaseManager): pass
...
>>> QueueManager.register('get_queue', callable=lambda: queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()
17.2.2.8. 代理对象¶
代理是一个 指向 其他共享对象的对象，这个对象(很可能)在另外一个进程中。共享对象也可以说是代理 指涉 的对象。多个代理对象可能指向同一个指涉对象。
A proxy object has methods which invoke corresponding methods of its referent (although not every method of the referent will necessarily be available through the proxy). A proxy can usually be used in most of the same ways that its referent can:
>>> from multiprocessing import Manager
>>> manager = Manager()
>>> l = manager.list([i*i for i in range(10)])
>>> print(l)
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
>>> print(repr(l))
<ListProxy object, typeid 'list' at 0x...>
>>> l[4]
16
>>> l[2:5]
[4, 9, 16]
注意，对代理使用 str() 函数会返回指涉对象的字符串表示，但是 repr() 却会返回代理本身的内部字符串表示。
An important feature of proxy objects is that they are picklable so they can be passed between processes. Note, however, that if a proxy is sent to the corresponding manager’s process then unpickling it will produce the referent itself. This means, for example, that one shared object can contain a second:
>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b) # referent of a now contains referent of b
>>> print(a, b)
[[]] []
>>> b.append('hello')
>>> print(a, b)
[['hello']] ['hello']
注解
multiprocessing 中的代理类并没有提供任何对于代理值比较的支持。所以，我们会得到如下结果:
>>> manager.list([1,2,3]) == [1,2,3]
False
当需要比较值的时候，应该替换为使用指涉对象的拷贝。
-
class
multiprocessing.managers.BaseProxy¶ 代理对象是
BaseProxy派生类的实例。-
_callmethod(methodname[, args[, kwds]])¶ 调用指涉对象的方法并返回结果。
如果
proxy是一个代理且其指涉的是obj, 那么下面的表达式:proxy._callmethod(methodname, args, kwds)
相当于求取以下表达式的值:
getattr(obj, methodname)(*args, **kwds)
于管理器进程。
返回结果会是一个值拷贝或者一个新的共享对象的代理 - 见函数
BaseManager.register()中关于参数 method_to_typeid 的文档。如果这个调用熬出了异常，则这个异常会被
_callmethod()透传出来。如果是管理器进程本身抛出的一些其他异常，则会被_callmethod()转换为RemoteError异常重新抛出。特别注意，如果 methodname 没有 暴露 出来，将会引发一个异常。
_callmethod()的一个使用示例:>>> l = manager.list(range(10)) >>> l._callmethod('__len__') 10 >>> l._callmethod('__getitem__', (slice(2, 7),)) # equivalent to l[2:7] [2, 3, 4, 5, 6] >>> l._callmethod('__getitem__', (20,)) # equivalent to l[20] Traceback (most recent call last): ... IndexError: list index out of range
-
_getvalue()¶ 返回指涉对象的一份拷贝。
如果指涉对象无法序列化，则会抛出一个异常。
-
__repr__()¶ 返回代理对象的字符串表示。
-
__str__()¶ 返回指涉对象的字符串表示。
-
17.2.2.9. 进程池¶
可以创建一个进程池，它将使用 Pool 类执行提交给它的任务。
-
class
multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])¶ 一个进程池对象，它控制可以提交作业的工作进程池。它支持带有超时和回调的异步结果，以及一个并行的 map 实现。
processes 是要使用的工作进程数目。如果 processes 为
None，则使用os.cpu_count()返回的值。如果 initializer 不为
None，则每个工作进程将会在启动时调用initializer(*initargs)。maxtasksperchild 是一个工作进程在它退出或被一个新的工作进程代替之前能完成的任务数量，为了释放未使用的资源。默认的 maxtasksperchild 是
None，意味着工作进程寿与池齐。context 可被用于指定启动的工作进程的上下文。通常一个进程池是使用函数
multiprocessing.Pool()或者一个上下文对象的Pool()方法创建的。在这两种情况下， context 都是适当设置的。注意，进程池对象的方法只有创建它的进程能够调用。
3.2 新版功能: maxtasksperchild
3.4 新版功能: context
注解
通常来说，
Pool中的 Worker 进程的生命周期和进程池的工作队列一样长。一些其他系统中（如 Apache, mod_wsgi 等）也可以发现另一种模式，他们会让工作进程在完成一些任务后退出，清理、释放资源，然后启动一个新的进程代替旧的工作进程。Pool的 maxtasksperchild 参数给用户提供了这种能力。-
apply(func[, args[, kwds]])¶ 使用 args 参数以及 kwds 命名参数调用 func , 它会返回结果前阻塞。这种情况下，
apply_async()更适合并行化工作。另外 func 只会在一个进程池中的一个工作进程中执行。
-
apply_async(func[, args[, kwds[, callback[, error_callback]]]])¶ apply()方法的一个变种，返回一个结果对象。如果指定了 callback , 它必须是一个接受单个参数的可调用对象。当执行成功时， callback 会被用于处理执行后的返回结果，否则，调用 error_callback 。
如果指定了 error_callback , 它必须是一个接受单个参数的可调用对象。当目标函数执行失败时， 会将抛出的异常对象作为参数传递给 error_callback 执行。
回调函数应该立即执行完成，否则会阻塞负责处理结果的线程。
-
map(func, iterable[, chunksize])¶ A parallel equivalent of the
map()built-in function (it supports only one iterable argument though). It blocks until the result is ready.这个方法会将可迭代对象分割为许多块，然后提交给进程池。可以将 chunksize 设置为一个正整数从而（近似）指定每个块的大小可以。
-
map_async(func, iterable[, chunksize[, callback[, error_callback]]])¶ 和
map()方法类似，但是返回一个结果对象。如果指定了 callback , 它必须是一个接受单个参数的可调用对象。当执行成功时， callback 会被用于处理执行后的返回结果，否则，调用 error_callback 。
如果指定了 error_callback , 它必须是一个接受单个参数的可调用对象。当目标函数执行失败时， 会将抛出的异常对象作为参数传递给 error_callback 执行。
回调函数应该立即执行完成，否则会阻塞负责处理结果的线程。
-
imap(func, iterable[, chunksize])¶ A lazier version of
map().chunksize 参数的作用和
map()方法的一样。对于很长的迭代器，给 chunksize 设置一个很大的值会比默认值1极大 地加快执行速度。同样，如果 chunksize 是
1, 那么imap()方法所返回的迭代器的next()方法拥有一个可选的 timeout 参数： 如果无法在 timeout 秒内执行得到结果，则``next(timeout)`` 会抛出multiprocessing.TimeoutError异常。
-
imap_unordered(func, iterable[, chunksize])¶ 和
imap()相同，只不过通过迭代器返回的结果是任意的。（当进程池中只有一个工作进程的时候，返回结果的顺序才能认为是”有序”的）
-
starmap(func, iterable[, chunksize])¶ 和
map()类似，不过 iterable 中的每一项会被解包再作为函数参数。比如可迭代对象
[(1,2), (3, 4)]会转化为等价于[func(1,2), func(3,4)]的调用。3.3 新版功能.
-
starmap_async(func, iterable[, chunksize[, callback[, error_back]]])¶ 相当于
starmap()与map_async()的结合，迭代 iterable 的每一项，解包作为 func 的参数并执行，返回用于获取结果的对象。3.3 新版功能.
-
close()¶ 阻止后续任务提交到进程池，当所有任务执行完成后，工作进程会退出。
-
terminate()¶ 不必等待未完成的任务，立即停止工作进程。当进程池对象呗垃圾回收时，
terminate()会立即调用。
-
join()¶ 等待工作进程结束。调用
join()前必须先调用close()或者terminate()。
3.3 新版功能: 进程池对象现在支持上下文管理器协议 - 参见 上下文管理器类型 。
__enter__()返回进程池对象,__exit__()会调用terminate()。-
-
class
multiprocessing.pool.AsyncResult¶ Pool.apply_async()和Pool.map_async()返回对象所属的类。-
get([timeout])¶ 用于获取执行结果。如果 timeout 不是
None并且在 timeout 秒内仍然没有执行完得到结果，则抛出multiprocessing.TimeoutError异常。如果远程调用发生异常，这个异常会通过get()重新抛出。
-
wait([timeout])¶ 阻塞，直到返回结果，或者 timeout 秒后超时。
-
ready()¶ 用于判断执行状态，是否已经完成。
-
successful()¶ Return whether the call completed without raising an exception. Will raise
AssertionErrorif the result is not ready.
-
下面的例子演示了进程池的用法:
from multiprocessing import Pool
import time
def f(x):
return x*x
if __name__ == '__main__':
with Pool(processes=4) as pool: # start 4 worker processes
result = pool.apply_async(f, (10,)) # evaluate "f(10)" asynchronously in a single process
print(result.get(timeout=1)) # prints "100" unless your computer is *very* slow
print(pool.map(f, range(10))) # prints "[0, 1, 4,..., 81]"
it = pool.imap(f, range(10))
print(next(it)) # prints "0"
print(next(it)) # prints "1"
print(it.next(timeout=1)) # prints "4" unless your computer is *very* slow
result = pool.apply_async(time.sleep, (10,))
print(result.get(timeout=1)) # raises multiprocessing.TimeoutError
17.2.2.10. 监听者及客户端¶
Usually message passing between processes is done using queues or by using
Connection objects returned by
Pipe().
不过，multiprocessing.connection 其实提供了一些更灵活的特性。最基础的用法是通过它抽象出来的高级来操作socket或者Windows命名管道。也提供一些高级用法，如通过 hmac 模块来支持 摘要认证，以及同时监听多个管道连接。
-
multiprocessing.connection.deliver_challenge(connection, authkey)¶ 发送一个随机生成的消息到另一端，并等待回复。
如果收到的回复与使用 authkey 生成的信息摘要匹配成功，就会发送一个欢迎信息给管道另一端。否则抛出
AuthenticationError异常。
-
multiprocessing.connection.answer_challenge(connection, authkey)¶ 接收一条信息，使用 authkey 作为键计算信息摘要，然后将摘要发送回去。
如果没有收到欢迎消息，就抛出
AuthenticationError异常。
-
multiprocessing.connection.Client(address[, family[, authenticate[, authkey]]])¶ Attempt to set up a connection to the listener which is using address address, returning a
Connection.连接的类型取决于 family 参数，但是通常可以省略，因为可以通过 address 的格式推导出来。(查看 地址格式 )
If authenticate is
Trueor authkey is a byte string then digest authentication is used. The key used for authentication will be either authkey orcurrent_process().authkeyif authkey isNone. If authentication fails thenAuthenticationErroris raised. See 认证密码.
-
class
multiprocessing.connection.Listener([address[, family[, backlog[, authenticate[, authkey]]]]])¶ 可以监听连接请求，是对于绑定套接字或者 Windows 命名管道的封装。
address 是监听器对象中的绑定套接字或命名管道使用的地址。
注解
如果使用 ‘0.0.0.0’ 作为监听地址，那么在Windows上这个地址无法建立连接。想要建立一个可连接的端点，应该使用 ‘127.0.0.1’ 。
family 是套接字(或者命名管道)使用的类型。它可以是以下一种:
'AF_INET'( TCP 套接字类型),'AF_UNIX'( Unix 域套接字) 或者'AF_PIPE'( Windows 命名管道)。其中只有第一个保证各平台可用。如果 family 是None,那么 family 会根据 address 的格式自动推导出来。如果 address 也是None, 则取默认值。默认值为可用类型中速度最快的。见 地址格式 。注意，如果 family 是'AF_UNIX'而address是``None`` ,套接字会在一个tempfile.mkstemp()创建的私有临时目录中创建。如果监听器对象使用了套接字，backlog (默认值为1) 会在套接字绑定后传递给它的
listen()方法。If authenticate is
True(Falseby default) or authkey is notNonethen digest authentication is used.If authkey is a byte string then it will be used as the authentication key; otherwise it must be
None.If authkey is
Noneand authenticate isTruethencurrent_process().authkeyis used as the authentication key. If authkey isNoneand authenticate isFalsethen no authentication is done. If authentication fails thenAuthenticationErroris raised. See 认证密码.-
accept()¶ Accept a connection on the bound socket or named pipe of the listener object and return a
Connectionobject. If authentication is attempted and fails, thenAuthenticationErroris raised.
-
close()¶ 关闭监听器上的绑定套接字或者命名管道。此函数会在监听器被垃圾回收后自动调用。不过仍然建议显式调用函数关闭。
监听器对象拥有下列只读属性:
-
address¶ 被监听器对象使用的地址。
-
last_accepted¶ 最后一个连接所使用的地址。如果没有的话就是
None。
3.3 新版功能: 监听器对象现在支持了上下文管理协议 - 见 上下文管理器类型 。
__enter__()返回一个监听器对象,__exit__()会调用close()。-
-
multiprocessing.connection.wait(object_list, timeout=None)¶ 一直等待直到 object_list 中某个对象处于就绪状态。返回 object_list 中处于就绪状态的对象。如果 timeout 是一个浮点型，该方法会最多阻塞这么多秒。如果 timeout 是
None，则会允许阻塞的事件没有限制。timeout为负数的情况下和为0的情况相同。对于 Unix 和 Windows ，下列对象都可以出现在 object_list 中
- a readable
Connectionobject; - 一个已连接并且可读的
socket.socket对象；或者 Process对象中的sentinel属性。
当一个连接或者套接字对象拥有有效的数据可被读取的时候，或者另一端关闭后，这个对象就处于就绪状态。
Unix:
wait(object_list, timeout)和select.select(object_list, [], [], timeout)几乎相同。差别在于，如果select.select()被信号中断，它会抛出一个附带错误号为EINTR的OSError异常，而wait()不会。Windows: object_list 中的元素必须是一个表示为整数的可等待的句柄(按照 Win32 函数
WaitForMultipleObjects()的文档中所定义) 或者一个拥有fileno()方法的对象，这个对象返回一个套接字句柄或者管道句柄。（注意管道和套接字两种句柄 不是 可等待的句柄）3.3 新版功能.
- a readable
示例
下面的服务代码创建了一个使用 'secret password' 作为认证密码的监听器。它会等待连接然后发送一些数据给客户端:
from multiprocessing.connection import Listener
from array import array
address = ('localhost', 6000) # family is deduced to be 'AF_INET'
with Listener(address, authkey=b'secret password') as listener:
with listener.accept() as conn:
print('connection accepted from', listener.last_accepted)
conn.send([2.25, None, 'junk', float])
conn.send_bytes(b'hello')
conn.send_bytes(array('i', [42, 1729]))
下面的代码连接到服务然后从服务器上j接收一些数据:
from multiprocessing.connection import Client
from array import array
address = ('localhost', 6000)
with Client(address, authkey=b'secret password') as conn:
print(conn.recv()) # => [2.25, None, 'junk', float]
print(conn.recv_bytes()) # => 'hello'
arr = array('i', [0, 0, 0, 0, 0])
print(conn.recv_bytes_into(arr)) # => 8
print(arr) # => array('i', [42, 1729, 0, 0, 0])
下面的代码使用了 wait() ，以便在同时等待多个进程发来消息。
import time, random
from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait
def foo(w):
for i in range(10):
w.send((i, current_process().name))
w.close()
if __name__ == '__main__':
readers = []
for i in range(4):
r, w = Pipe(duplex=False)
readers.append(r)
p = Process(target=foo, args=(w,))
p.start()
# We close the writable end of the pipe now to be sure that
# p is the only process which owns a handle for it. This
# ensures that when p closes its handle for the writable end,
# wait() will promptly report the readable end as being ready.
w.close()
while readers:
for r in wait(readers):
try:
msg = r.recv()
except EOFError:
readers.remove(r)
else:
print(msg)
17.2.2.11. 认证密码¶
When one uses Connection.recv, the
data received is automatically
unpickled. Unfortunately unpickling data from an untrusted source is a security
risk. Therefore Listener and Client() use the hmac module
to provide digest authentication.
认证密钥是一个 byte 类型的字符串，可以认为是和密码一样的东西，连接建立好后，双方都会要求另一方证明知道认证密钥。（这个证明过程不会通过连接发送密钥）
如果要求认证但是没有指定认证密钥，则会使用 current_process().authkey 的返回值 (参见 Process)。 这个值将被当前进程所创建的任何 Process 对象自动继承。 这意味着 (在默认情况下) 一个包含多进程的程序中的所有进程会在相互间建立连接的时候共享单个认证密钥。
os.urandom() 也可以用来生成合适的认证密钥。
17.2.2.12. 日志记录¶
当前模块也提供了一些对 logging 的支持。注意， logging 模块本身并没有使用进程间共享的锁，所以来自于多个进程的日志可能（具体取决于使用的日志 handler）相互覆盖或者混杂。
-
multiprocessing.get_logger()¶ 返回
multiprocessing使用的 logger，必要的话会创建一个新的。如果创建的首个 logger 日志级别为
logging.NOTSET并且没有默认 handler。通过这个 logger 打印的消息不会传递到根 logger。注意在 Windows 上，子进程只会继承父进程 logger 的日志级别 - 对于logger的其他自定义项不会继承。
-
multiprocessing.log_to_stderr()¶ 此函数会调用
get_logger()但是会在返回的 logger 上增加一个 handler，将所有输出都使用'[%(levelname)s/%(processName)s] %(message)s'的格式发送到sys.stderr。
下面是一个在交互式解释器中打开日志功能的例子:
>>> import multiprocessing, logging
>>> logger = multiprocessing.log_to_stderr()
>>> logger.setLevel(logging.INFO)
>>> logger.warning('doomed')
[WARNING/MainProcess] doomed
>>> m = multiprocessing.Manager()
[INFO/SyncManager-...] child process calling self.run()
[INFO/SyncManager-...] created temp directory /.../pymp-...
[INFO/SyncManager-...] manager serving at '/.../listener-...'
>>> del m
[INFO/MainProcess] sending shutdown message to manager
[INFO/SyncManager-...] manager exiting with exitcode 0
要查看日志等级的完整列表，见 logging 模块。
17.2.2.13. multiprocessing.dummy 模块¶
multiprocessing.dummy 复制了 multiprocessing 的 API，不过是在 threading 模块之上包装了一层。
17.2.3. 编程指导¶
使用 multiprocessing 时，应遵循一些指导原则和习惯用法。
17.2.3.1. 所有启动方法¶
下面这些使用于所有启动方法。
避免共享状态
应该尽可能避免在进程间传递大量数据，越少越好。
最好坚持使用队列或者管道进行进程间通信，而不是底层的同步原语。
可序列化
保证所代理的方法的参数是可以序列化的。
代理的线程安全性
不要在多线程中同时使用一个代理对象，除非你用锁保护它。
（而在不同进程中使用 相同 的代理对象从不会发生问题。）
使用 Join 避免僵尸进程
在 Unix 上，如果一个进程执行完成但是没有被 join，就会变成僵尸进程。一般来说，僵尸进程不会很多，因为每次新启动进程（或者active_children()被调用）时，所有已执行完成且没有被 join 的进程都会自动被 join，而且对一个执行完的进程调用Process.is_alive也会 join 这个进程。尽管如此，对自己启动的进程显式调用 join 依然是最佳实践。
继承优于序列化、反序列化
当使用 spawn 或者 forkserver 的启动方式时，multiprocessing中的许多类型都必须是可序列化的，这样子进程才能使用它们。但是通常我们都应该避免使用管道和队列发送共享对象到另外一个进程，而是重新组织代码，对于其他进程创建出来的共享对象，让那些需要访问这些对象的子进程可以直接将这些对象从父进程继承过来。
避免杀死进程
听过
Process.terminate停止一个进程很容易导致这个进程正在使用的共享资源（如锁、信号量、管道和队列）损坏或者变得不可用，无法在其他进程中继续使用。所以，最好只对那些从来不使用共享资源的进程调用
Process.terminate。
Join 使用队列的进程
记住，往队列放入数据的进程会一直等待直到所有缓存项被”feeder” 线程传给底层管道。（子进程可以调用队列的
Queue.cancel_join_thread方法禁止这种行为）这意味着，任何使用队列的时候，你都要确保在进程join之前，所有存放到队列中的项将会被其他进程、线程完全消费。否则不能保证这个写过队列的进程可以正常终止。记住非精灵进程会自动 join 。
下面是一个会导致死锁的例子:
from multiprocessing import Process, Queue def f(q): q.put('X' * 1000000) if __name__ == '__main__': queue = Queue() p = Process(target=f, args=(queue,)) p.start() p.join() # this deadlocks obj = queue.get()交换最后两行可以修复这个问题（或者直接删掉
p.join()）。
显示传递资源给子进程
在Unix上，使用 fork 方式启动的子进程可以使用父进程中全局创建的共享资源。
除了（部分原因）让代码兼容 Windows 以及其他的进程启动方式外，这种形式还保证了在子进程生命期这个对象是不会被父进程垃圾回收的。如果父进程中的某些对象被垃圾回收会导致资源释放，这就变得很重要。
所以对于实例：
from multiprocessing import Process, Lock def f(): ... do something using "lock" ... if __name__ == '__main__': lock = Lock() for i in range(10): Process(target=f).start()应当重写成这样：
from multiprocessing import Process, Lock def f(l): ... do something using "l" ... if __name__ == '__main__': lock = Lock() for i in range(10): Process(target=f, args=(lock,)).start()
谨防将 sys.stdin 数据替换为 “类似文件的对象”
multiprocessing内部会无条件地这样调用:os.close(sys.stdin.fileno())在
multiprocessing.Process._bootstrap()方法中 —— 这会导致与”进程中的进程”相关的一些问题。这已经被修改成了:sys.stdin.close() sys.stdin = open(os.open(os.devnull, os.O_RDONLY), closefd=False)它解决了进程相互冲突导致文件描述符错误的根本问题，但是对使用带缓冲的“文件类对象”替换
sys.stdin()作为输出的应用程序造成了潜在的危险。如果多个进程调用了此文件类对象的close()方法，会导致相同的数据多次刷写到此对象，损坏数据。如果你写入文件类对象并实现了自己的缓存，可以在每次追加缓存数据时记录当前进程id，从而将其变成 fork 安全的，当发现进程id变化后舍弃之前的缓存，例如:
@property def cache(self): pid = os.getpid() if pid != self._pid: self._pid = pid self._cache = [] return self._cache
17.2.3.2. spawn 和 forkserver 启动方式¶
相对于 fork 启动方式，有一些额外的限制。
更依赖序列化
全局变量
记住，如果子进程中的代码尝试访问一个全局变量，它所看到的值可能和父进程中执行
Process.start那一刻的值不一样。当全局变量知识模块级别的常量时，是不会有问题的。
安全导入主模块
确保主模块可以被新启动的Python解释器安全导入而不会引发什么副作用（比如又启动了一个子进程）
例如，使用 spawn 或 forkserver 启动方式执行下面的模块，会引发
RuntimeError异常而失败。from multiprocessing import Process def foo(): print('hello') p = Process(target=foo) p.start()应该通过下面的方法使用
if __name__ == '__main__':，从而保护程序”入口点”:from multiprocessing import Process, freeze_support, set_start_method def foo(): print('hello') if __name__ == '__main__': freeze_support() set_start_method('spawn') p = Process(target=foo) p.start()（如果程序将正常运行而不是冻结，则可以省略
freeze_support()行）这允许新启动的 Python 解释器安全导入模块然后运行模块中的
foo()函数。如果主模块中创建了进程池或者管理器，这个规则也适用。
17.2.4. 例子¶
创建和使用自定义管理器、代理的示例:
from multiprocessing import freeze_support
from multiprocessing.managers import BaseManager, BaseProxy
import operator
##
class Foo:
def f(self):
print('you called Foo.f()')
def g(self):
print('you called Foo.g()')
def _h(self):
print('you called Foo._h()')
# A simple generator function
def baz():
for i in range(10):
yield i*i
# Proxy type for generator objects
class GeneratorProxy(BaseProxy):
_exposed_ = ['__next__']
def __iter__(self):
return self
def __next__(self):
return self._callmethod('__next__')
# Function to return the operator module
def get_operator_module():
return operator
##
class MyManager(BaseManager):
pass
# register the Foo class; make `f()` and `g()` accessible via proxy
MyManager.register('Foo1', Foo)
# register the Foo class; make `g()` and `_h()` accessible via proxy
MyManager.register('Foo2', Foo, exposed=('g', '_h'))
# register the generator function baz; use `GeneratorProxy` to make proxies
MyManager.register('baz', baz, proxytype=GeneratorProxy)
# register get_operator_module(); make public functions accessible via proxy
MyManager.register('operator', get_operator_module)
##
def test():
manager = MyManager()
manager.start()
print('-' * 20)
f1 = manager.Foo1()
f1.f()
f1.g()
assert not hasattr(f1, '_h')
assert sorted(f1._exposed_) == sorted(['f', 'g'])
print('-' * 20)
f2 = manager.Foo2()
f2.g()
f2._h()
assert not hasattr(f2, 'f')
assert sorted(f2._exposed_) == sorted(['g', '_h'])
print('-' * 20)
it = manager.baz()
for i in it:
print('<%d>' % i, end=' ')
print()
print('-' * 20)
op = manager.operator()
print('op.add(23, 45) =', op.add(23, 45))
print('op.pow(2, 94) =', op.pow(2, 94))
print('op._exposed_ =', op._exposed_)
##
if __name__ == '__main__':
freeze_support()
test()
使用 Pool:
import multiprocessing
import time
import random
import sys
#
# Functions used by test code
#
def calculate(func, args):
result = func(*args)
return '%s says that %s%s = %s' % (
multiprocessing.current_process().name,
func.__name__, args, result
)
def calculatestar(args):
return calculate(*args)
def mul(a, b):
time.sleep(0.5 * random.random())
return a * b
def plus(a, b):
time.sleep(0.5 * random.random())
return a + b
def f(x):
return 1.0 / (x - 5.0)
def pow3(x):
return x ** 3
def noop(x):
pass
#
# Test code
#
def test():
PROCESSES = 4
print('Creating pool with %d processes\n' % PROCESSES)
with multiprocessing.Pool(PROCESSES) as pool:
#
# Tests
#
TASKS = [(mul, (i, 7)) for i in range(10)] + \
[(plus, (i, 8)) for i in range(10)]
results = [pool.apply_async(calculate, t) for t in TASKS]
imap_it = pool.imap(calculatestar, TASKS)
imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)
print('Ordered results using pool.apply_async():')
for r in results:
print('\t', r.get())
print()
print('Ordered results using pool.imap():')
for x in imap_it:
print('\t', x)
print()
print('Unordered results using pool.imap_unordered():')
for x in imap_unordered_it:
print('\t', x)
print()
print('Ordered results using pool.map() --- will block till complete:')
for x in pool.map(calculatestar, TASKS):
print('\t', x)
print()
#
# Test error handling
#
print('Testing error handling:')
try:
print(pool.apply(f, (5,)))
except ZeroDivisionError:
print('\tGot ZeroDivisionError as expected from pool.apply()')
else:
raise AssertionError('expected ZeroDivisionError')
try:
print(pool.map(f, list(range(10))))
except ZeroDivisionError:
print('\tGot ZeroDivisionError as expected from pool.map()')
else:
raise AssertionError('expected ZeroDivisionError')
try:
print(list(pool.imap(f, list(range(10)))))
except ZeroDivisionError:
print('\tGot ZeroDivisionError as expected from list(pool.imap())')
else:
raise AssertionError('expected ZeroDivisionError')
it = pool.imap(f, list(range(10)))
for i in range(10):
try:
x = next(it)
except ZeroDivisionError:
if i == 5:
pass
except StopIteration:
break
else:
if i == 5:
raise AssertionError('expected ZeroDivisionError')
assert i == 9
print('\tGot ZeroDivisionError as expected from IMapIterator.next()')
print()
#
# Testing timeouts
#
print('Testing ApplyResult.get() with timeout:', end=' ')
res = pool.apply_async(calculate, TASKS[0])
while 1:
sys.stdout.flush()
try:
sys.stdout.write('\n\t%s' % res.get(0.02))
break
except multiprocessing.TimeoutError:
sys.stdout.write('.')
print()
print()
print('Testing IMapIterator.next() with timeout:', end=' ')
it = pool.imap(calculatestar, TASKS)
while 1:
sys.stdout.flush()
try:
sys.stdout.write('\n\t%s' % it.next(0.02))
except StopIteration:
break
except multiprocessing.TimeoutError:
sys.stdout.write('.')
print()
print()
if __name__ == '__main__':
multiprocessing.freeze_support()
test()
一个演示如何使用队列来向一组工作进程提供任务并收集结果的例子：
import time
import random
from multiprocessing import Process, Queue, current_process, freeze_support
#
# Function run by worker processes
#
def worker(input, output):
for func, args in iter(input.get, 'STOP'):
result = calculate(func, args)
output.put(result)
#
# Function used to calculate result
#
def calculate(func, args):
result = func(*args)
return '%s says that %s%s = %s' % \
(current_process().name, func.__name__, args, result)
#
# Functions referenced by tasks
#
def mul(a, b):
time.sleep(0.5*random.random())
return a * b
def plus(a, b):
time.sleep(0.5*random.random())
return a + b
#
#
#
def test():
NUMBER_OF_PROCESSES = 4
TASKS1 = [(mul, (i, 7)) for i in range(20)]
TASKS2 = [(plus, (i, 8)) for i in range(10)]
# Create queues
task_queue = Queue()
done_queue = Queue()
# Submit tasks
for task in TASKS1:
task_queue.put(task)
# Start worker processes
for i in range(NUMBER_OF_PROCESSES):
Process(target=worker, args=(task_queue, done_queue)).start()
# Get and print results
print('Unordered results:')
for i in range(len(TASKS1)):
print('\t', done_queue.get())
# Add more tasks using `put()`
for task in TASKS2:
task_queue.put(task)
# Get and print some more results
for i in range(len(TASKS2)):
print('\t', done_queue.get())
# Tell child processes to stop
for i in range(NUMBER_OF_PROCESSES):
task_queue.put('STOP')
if __name__ == '__main__':
freeze_support()
test()