Python标准库multiprocessing基于进程的并行

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 参数中获取顺序和关键字参数。

start()¶

启动进程活动。

每个进程对象最多只能调用一次。它安排对象的 run() 方法在一个单独的进程中调用。

join([timeout])¶

如果可选参数 timeout 是 None （默认值），则该方法将阻塞，直到调用 join() 方法的进程终止。如果 timeout 是一个正数，它最多会阻塞 timeout 秒。请注意，如果进程终止或方法超时，则该方法返回 None 。检查进程的 exitcode 以确定它是否终止。

一个进程可以合并多次。

进程无法并入自身，因为这会导致死锁。尝试在启动进程之前合并进程是错误的。

name¶

进程的名称。该名称是一个字符串，仅用于识别目的。它没有语义。可以为多个进程指定相同的名称。

初始名称由构造器设定。 如果没有为构造器提供显式名称，则会构造一个形式为 'Process-N₁:N₂:...:N_k' 的名称，其中每个 N_k 是其父亲的第 N 个孩子。

is_alive()¶

返回进程是否还活着。

粗略地说，从 start() 方法返回到子进程终止之前，进程对象仍处于活动状态。

daemon¶

进程的守护标志，一个布尔值。这必须在 start() 被调用之前设置。

初始值继承自创建进程。

当进程退出时，它会尝试终止其所有守护进程子进程。

请注意，不允许在守护进程中创建子进程。这是因为当守护进程由于父进程退出而中断时，其子进程会变成孤儿进程。 另外，这些 不是 Unix 守护进程或服务，它们是正常进程，如果非守护进程已经退出，它们将被终止（并且不被合并）。

除了 threading.Thread API ，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子句等。

请注意，进程的后代进程将不会被终止 —— 它们将简单地变成孤立的。

警告

如果在关联进程使用管道或队列时使用此方法，则管道或队列可能会损坏，并可能无法被其他进程使用。类似地，如果进程已获得锁或信号量等，则终止它可能导致其他进程死锁。

kill()¶

与 terminate() 相同，但在Unix上使用 SIGKILL 信号。

3.7 新版功能.

close()¶

关闭 Process 对象，释放与之关联的所有资源。如果底层进程仍在运行，则会引发 ValueError 。一旦 close() 成功返回， Process 对象的大多数其他方法和属性将引发 ValueError 。

3.7 新版功能.

注意 start() 、 join() 、 is_alive() 、 terminate() 和 exitcode 方法只能由创建进程对象的进程调用。

Process 一些方法的示例用法：

>>> importmultiprocessing,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¶

有超时的方法超时时引发。

管道和队列¶

使用多进程时，一般使用消息机制实现进程间通信，尽可能避免使用同步原语，例如锁。

消息机制包含： Pipe() (可以用于在两个进程间传递消息)，以及队列(能够在多个生产者和消费者之间通信)。

Queue, SimpleQueue 以及 JoinableQueue 都是多生产者，多消费者，并且实现了 FIFO 的队列类型，其表现与标准库中的 queue.Queue 类相似。 不同之处在于 Queue 缺少标准库的 queue.Queue 从 Python 2.5 开始引入的 task_done() 和 join() 方法。

如果你使用了 JoinableQueue ，那么你 必须 对每个已经移出队列的任务调用 JoinableQueue.task_done()。 不然的话用于统计未完成任务的信号量最终会溢出并抛出异常。

另外还可以通过使用一个管理器对象创建一个共享队列，详见 数据管理器 。

该 例子 展示了如何使用队列实现进程间通信。

multiprocessing.Pipe([duplex])¶

返回一对 Connection`对象  ``(conn1,conn2)` ， 分别表示管道的两端。

如果 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() 阻塞被解除。

杂项¶

multiprocessing.active_children()¶

返回当前进程存活的子进程的列表。

调用该方法有“等待”已经结束的进程的副作用。

multiprocessing.cpu_count()¶

返回系统的CPU数量。

该数量不同于当前进程可以使用的CPU数量。可用的CPU数量可以由 len(os.sched_getaffinity(0)) 方法获得。

可能引发 NotImplementedError 。

参见

os.cpu_count()

multiprocessing.current_process()¶

返回与当前进程相对应的 Process 对象。

和 threading.current_thread() 相同。

multiprocessing.freeze_support()¶

为使用了 multiprocessing  的程序，提供冻结以产生 Windows 可执行文件的支持。(在 py2exe, PyInstaller 和 cx_Freeze 上测试通过)

需要在 main 模块的 if__name__=='__main__' 该行之后马上调用该函数。例如：

frommultiprocessingimportProcess,freeze_support
deff():
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 新版功能.

连接对象（Connection）¶

Connection 对象允许收发可以序列化的对象或字符串。它们可以看作面向消息的连接套接字。

通常使用 Pipe 创建 Connection 对象。详见 ： 监听者及客户端.

class multiprocessing.connection.Connection¶

send(obj)¶

将一个对象发送到连接的另一端，可以用 recv() 读取。

发送的对象必须是可以序列化的，过大的对象 ( 接近 32MiB+ ，这个值取决于操作系统 ) 有可能引发 ValueError  异常。

recv()¶

返回一个由另一端使用 send() 发送的对象。该方法会一直阻塞直到接收到对象。 如果对端关闭了连接或者没有东西可接收，将抛出 EOFError  异常。

fileno()¶

返回由连接对象使用的描述符或者句柄。

close()¶

关闭连接对象。

当连接对象被垃圾回收时会自动调用。

poll([timeout])¶

返回连接对象中是否有可以读取的数据。

如果未指定 timeout ，此方法会马上返回。如果 timeout 是一个数字，则指定了最大阻塞的秒数。如果 timeout 是 None  ，那么将一直等待，不会超时。

注意通过使用 multiprocessing.connection.wait() 可以一次轮询多个连接对象。

send_bytes(buffer[, offset[, size]])¶

从一个 bytes-like object  （字节类对象）对象中取出字节数组并作为一条完整消息发送。

如果由 offset  给定了在 buffer 中读取数据的位置。 如果给定了 size ，那么将会从缓冲区中读取多个字节。 过大的缓冲区 ( 接近 32MiB+ ，此值依赖于操作系统 ) 有可能引发 ValueError  异常。

recv_bytes([maxlength])¶

以字符串形式返回一条从连接对象另一端发送过来的字节数据。此方法在接收到数据前将一直阻塞。 如果连接对象被对端关闭或者没有数据可读取，将抛出 EOFError  异常。

如果给定了 maxlength 并且消息长于 maxlength 那么将抛出 OSError 并且该连接对象将不再可读。

在 3.3 版更改: 曾经该函数抛出 IOError ，现在这是 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() 。

例如:

>>> frommultiprocessingimportPipe
>>> 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'
>>> importarray
>>> arr1=array.array('i',range(5))
>>> arr2=array.array('i',[0]*10)
>>> a.send_bytes(arr1)
>>> count=b.recv_bytes_into(arr2)
>>> assertcount==len(arr1)*arr1.itemsize
>>> arr2
array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])

同步原语¶

通常来说同步原语在多进程环境中并不像它们在多线程环境中那么必要。参考 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  对象。

Lock supports the context manager protocol and thus may be

used in with statements.

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 。

共享 ctypes 对象¶

在共享内存上创建可被子进程继承的共享对象时是可行的。

multiprocessing.Value(typecode_or_type, *args, lock=True)¶

返回一个从共享内存上创建的 ctypes 对象。默认情况下返回的实际上是经过了同步包装器包装过的。可以通过 Value 的 value 属性访问这个对象本身。

typecode_or_type 指明了返回的对象类型: 它可能是一个 ctypes 类型或者 array  模块中每个类型对应的单字符长度的字符串。 *args 会透传给这个类的构造函数。

如果 lock 参数是 True （默认值）, 将会新建一个递归锁用于同步对于此值的访问操作。 如果 lock 是 Lock 或者 RLock 对象，那么这个传入的锁将会用于同步对这个值的访问操作，如果 lock 是 False , 那么对这个对象的访问将没有锁保护，也就是说这个变量不是进程安全的。

诸如 += 这类的操作会引发独立的读操作和写操作，也就是说这类操作符并不具有原子性。所以，如果你想让递增操作具有原子性，这样的方式并不能达到要求:

counter.value+=1

假设共享对象内部关联的锁时递归锁(默认情况下就是)， 那么你可以采用这种方式

withcounter.get_lock():
counter.value+=1

注意 locl 只能是命名参数。

multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True)¶

从共享内存中申请并返回一个具有ctypes类型的数组对象。默认情况下返回值实际上是被同步器包装过的数组对象。

typecode_or_type 指明了返回的数组中的元素类型: 它可能是一个 ctypes 类型或者 array 模块中每个类型对应的单字符长度的字符串。 如果 size_or_initializer 是一个整数，那就会当做数组的长度，并且整个数组的内存会初始化为0。否则，如果 size_or_initializer 会被当成一个序列用于初始化数组中的内一个元素，并且会根据元素个数自动判断数组的长度。

如果 lock 为 True (默认值) 则将创建一个新的锁对象用于同步对值的访问。 如果 lock 为一个 Lock 或 RLock 对象则该对象将被用于同步对值的访问。 如果 lock 为 False 则对返回对象的访问将不会自动得到锁的保护，也就是说它不是“进程安全的”。

请注意 lock 是一个仅限关键字参数。

请注意 ctypes.c_char 的数组具有 value 和 raw 属性，允许被用来保存和提取字符串。

frommultiprocessingimportProcess,Lock
frommultiprocessing.sharedctypesimportValue,Array
fromctypesimportStructure,c_double
classPoint(Structure):
_fields_=[('x',c_double),('y',c_double)]
defmodify(n,x,s,A):
n.value**=2
x.value**=2
s.value=s.value.upper()
forainA:
a.x**=2
a.y**=2
if__name__=='__main__':
lock=Lock()
n=Value('i',7)
x=Value(c_double,1.0/3.0,lock=False)
s=Array('c',b'hello world',lock=lock)
A=Array(Point,[(1.875,-6.25),(-5.75,2.0),(2.375,9.5)],lock=lock)
p=Process(target=modify,args=(n,x,s,A))
p.start()
p.join()
print(n.value)
print(x.value)
print(s.value)
print([(a.x,a.y)forainA])

49
0.1111111111111111
HELLO WORLD
[(3.515625, 39.0625), (33.0625, 4.0), (5.640625, 90.25)]

frommultiprocessing.managersimportBaseManager
classMathsClass:
defadd(self,x,y):
returnx+y
defmul(self,x,y):
returnx*y
classMyManager(BaseManager):
pass
MyManager.register('Maths',MathsClass)
if__name__=='__main__':
withMyManager()asmanager:
maths=manager.Maths()
print(maths.add(4,3))# prints 7
print(maths.mul(7,8))# prints 56

>>> frommultiprocessing.managersimportBaseManager
>>> fromqueueimportQueue
>>> queue=Queue()
>>> classQueueManager(BaseManager):pass
>>> QueueManager.register('get_queue',callable=lambda:queue)
>>> m=QueueManager(address=('',50000),authkey=b'abracadabra')
>>> s=m.get_server()
>>> s.serve_forever()

>>> frommultiprocessing.managersimportBaseManager
>>> classQueueManager(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')

>>> frommultiprocessing.managersimportBaseManager
>>> classQueueManager(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'

>>> frommultiprocessingimportProcess,Queue
>>> frommultiprocessing.managersimportBaseManager
>>> classWorker(Process):
... def__init__(self,q):
... self.q=q
... super(Worker,self).__init__()
... defrun(self):
... self.q.put('local hello')
...
>>> queue=Queue()
>>> w=Worker(queue)
>>> w.start()
>>> classQueueManager(BaseManager):pass
...
>>> QueueManager.register('get_queue',callable=lambda:queue)
>>> m=QueueManager(address=('',50000),authkey=b'abracadabra')
>>> s=m.get_server()
>>> s.serve_forever()

>>> frommultiprocessingimportManager
>>> manager=Manager()
>>> l=manager.list([i*iforiinrange(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]

>>> a=manager.list()
>>> b=manager.list()
>>> a.append(b)# referent of a now contains referent of b
>>> print(a,b)
[<ListProxy object, typeid 'list' at ...>] []
>>> b.append('hello')
>>> print(a[0],b)
['hello'] ['hello']

>>> l_outer=manager.list([manager.dict()foriinrange(2)])
>>> d_first_inner=l_outer[0]
>>> d_first_inner['a']=1
>>> d_first_inner['b']=2
>>> l_outer[1]['c']=3
>>> l_outer[1]['z']=26
>>> print(l_outer[0])
{'a': 1, 'b': 2}
>>> print(l_outer[1])
{'c': 3, 'z': 26}

# 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
# updating the dictionary, the proxy is notified of the change
lproxy[0]=d

>>> manager.list([1,2,3])==[1,2,3]
False

frommultiprocessingimportPool
importtime
deff(x):
returnx*x
if__name__=='__main__':
withPool(processes=4)aspool:# 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