Python多进程同步与通信机制详解：Lock、Queue与Pipe实战
本文深入解析Python中多进程同步与通信的核心技术。针对多线程无法充分利用多核的问题，重点讲解multiprocessing模块。通过实例演示如何使用Lock解决输出竞争问题，利用Semaphore管理资源数量，并详细对比Queue和Pipe两种进程间通信方式及其常用方法。最后介绍Manager实现复杂数据结构共享，帮助开发者高效构建并发程序。
Python中的多线程受全局解释器锁（GIL）限制，并非真正的多线程执行。若要充分利用多核CPU性能，在Python中通常需采用多进程方案。Python提供了功能完善的multiprocessing包，用户只需定义处理函数，库会自动完成其余工作。借助该包，可轻松实现从单进程到并发执行的转换。multiprocessing支持子进程创建、进程间通信与共享数据、多种同步机制，并提供了Process、Queue、Pipe、Lock等关键组件。
引例：
参考此前创建多进程的代码示例。
# -*- coding:utf-8 -*-
from multiprocessing import Process,Pool
import os,time
def run_proc(name): ##定义一个函数用于进程调用
for i in range(5):
time.sleep(0.2) #休眠0.2秒
print 'Run child process %s (%s)' % (name, os.getpid())
#执行一次该函数共需1秒的时间
if __name__ =='__main__': #执行主进程
print 'Run the main process (%s).' % (os.getpid())
mainStart = time.time() #记录主进程开始的时间
p = Pool(8) #开辟进程池
for i in range(16): #开辟14个进程
p.apply_async(run_proc,args=('Process'+str(i),))#每个进程都调用run_proc函数，
#args表示给该函数传递的参数。
print 'Waiting for all subprocesses done ...'
p.close() #关闭进程池
p.join() #等待开辟的所有进程执行完后，主进程才继续往下执行
print 'All subprocesses done'
mainEnd = time.time() #记录主进程结束时间
print 'All process ran %0.2f seconds.' % (mainEnd-mainStart) #主进程执行时间
运行结果：
Run the main process (36652). Waiting for all subprocesses done … Run child process Process0 (36708)Run child process Process1 (36748) Run child process Process3 (36736) Run child process Process2 (36716) Run child process Process4 (36768)
如第3行输出所示，偶尔会出现格式混乱的情况。这是由什么原因导致的呢？
原因在于多个进程争用打印输出资源。当一个进程尚未来得及输出换行符时，执行权可能切换至另一个进程，导致两个进程的输出内容合并在同一行。前一个进程的换行符需等到下次获得资源时才会打印，从而造成输出格式错乱。
Lock（锁）
为避免上述输出混乱，需在进程进入临界区（即访问临界资源的代码段）时加锁。添加锁后的执行效果如下所示：
# -*- coding:utf-8 -*-
lock = Lock() #申明一个全局的lock对象
def run_proc(name):
global lock #引用全局锁
for i in range(5):
time.sleep(0.2)
lock.acquire() #申请锁
print 'Run child process %s (%s)' % (name, os.getpid())
lock.release() #释放锁
Semaphore（信号量）
Semaphore用于信号量机制。当共享资源存在多个实例时，可使用Semaphore实现进程同步。其用法与Lock类似，通过s = Semaphore(N)初始化。每执行一次s.acquire()，可用资源数量减1；当资源数为0时，进程进入阻塞状态。每执行一次s.release()，释放占用的资源，可用数量加1。
多进程的通信（信息交互）
不同进程间进行数据交互时，初学者常尝试使用共享全局变量的方式，即通过读写全局变量实现信息传递。然而，这种方法在Python多进程中是行不通的。
通过以下示例，进一步加深对该问题的理解：
# -*- coding:utf-8 -*- from multiprocessing import Process, Pool import os import time L1 = [1, 2, 3] def add(a, b): global L1 L1 += range(a, b) print L1 if __name__ == '__main__': p1 = Process(target=add, args=(20, 30)) p2 = Process(target=add, args=(30, 40)) p1.start() p2.start() p1.join() p2.join() print L1
输出结果：
[1, 2, 3, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29] [1, 2, 3, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39] [1, 2, 3]
该程序初衷是将两个子进程生成的列表合并至全局变量L1中，但上述方法未能达到预期效果。既然无法通过全局变量实现进程间信息交互，那么可行的替代方案是什么？
multiprocessing模块提供了Queue和Pipe两种机制来实现进程间通信。
Queue（队列）
基于上述示例，使用Queue进行实现：
# -*- coding:utf-8 -*- from multiprocessing import Process, Queue, Lock L = [1, 2, 3] def add(q, lock, a, b): lock.acquire() # 加锁避免写入时出现不可预知的错误 L1 = range(a, b) lock.release() q.put(L1) print L1 if __name__ == '__main__': q = Queue() lock = Lock() p1 = Process(target=add, args=(q, lock, 20, 30)) p2 = Process(target=add, args=(q, lock, 30, 40)) p1.start() p2.start() p1.join() p2.join() L += q.get() + q.get() print L
执行结果：
[20, 21, 22, 23, 24, 25, 26, 27, 28, 29] [30, 31, 32, 33, 34, 35, 36, 37, 38, 39] [1, 2, 3, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
以下介绍Queue的常用方法：
- 定义队列时，可通过q = Queue(maxsize = 10)指定最大长度。若未指定或maxsize值小于1，队列默认为无限长度。
- q.put(item)方法用于向队列放入元素。该方法包含可选参数block，默认为True。此时若队列已满，进程将阻塞等待，直到有空闲位置。若block值为False，队列满时将抛出Full异常。
- Queue对象不可迭代，无法通过for循环直接取值，需每次调用q.get()方法获取。该方法同样包含可选参数block，默认为True。若队列为空，进程将阻塞等待。若block值为False，队列空时将抛出Empty异常。
- Queue.qsize() 返回当前队列中的元素数量。
- Queue.empty() 若队列为空返回True，否则返回False。
- Queue.full() 若队列已满返回True，否则返回False。
- Queue.get([block[, timeout]]) 获取队列元素，timeout指定等待时间。Queue.get_nowait() 等效于Queue.get(False)，即非阻塞模式。Queue.put(item) 写入队列，timeout指定等待时间。
- Queue.put_nowait(item) 等效于Queue.put(item, False)，即非阻塞模式。
Pipe（管道）
Pipe管道支持单向（half-duplex）或双向（duplex）通信。通过mutiprocessing.Pipe(duplex=False)可创建单向管道（默认创建双向管道）。双向Pipe允许两端进程既发送又接收数据；单向Pipe则规定前端端口用于接收，后端端口用于发送。
以下给出具体示例：
# -*- coding:utf-8 -*-
from multiprocessing import Process, Pipe
def proc1(pipe):
s = 'Hello,This is proc1'
pipe.send(s)
def proc2(pipe):
while True:
print "proc2 recieve:", pipe.recv()
if __name__ == "__main__":
pipe = Pipe()
p1 = Process(target=proc1, args=(pipe[0],))
p2 = Process(target=proc2, args=(pipe[1],))
p1.start()
p2.start()
p1.join()
p2.join(2) #限制执行时间最多为2秒
print '\nend all processes.'
执行结果如下：
proc2 recieve: Hello,This is proc1 proc2 recieve: end all processes.
当第二行输出后，由于管道中无新数据，Proc2处于阻塞状态，2秒后被强制结束。
以下是单向管道的示例，请注意pipe[0]和pipe[1]的角色分配。
# -*- coding:utf-8 -*-
from multiprocessing import Process, Pipe
def proc1(pipe):
s = 'Hello,This is proc1'
pipe.send(s)
def proc2(pipe):
while True:
print "proc2 recieve:", pipe.recv()
if __name__ == "__main__":
pipe = Pipe(duplex=False)
p1 = Process(target=proc1, args=(pipe[1],)) #pipe[1]为发送端
p2 = Process(target=proc2, args=(pipe[0],)) #pipe[0]为接收端
p1.start()
p2.start()
p1.join()
p2.join(2) # 限制执行时间最多为2秒
print '\nend all processes.'
执行结果与前述示例相同。
强大的Manager（管理器）
Queue和Pipe实现的数据共享仅支持Value和Array两种结构。Python提供了功能强大的Manager模块专门用于数据共享，其支持的类型非常丰富，包括：Value，Array，list, dict，Queue, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event等。
其具体用法如下：
from multiprocessing import Process, Manager
def func(dt, lt):
for i in range(10):
key = 'arg' + str(i)
dt[key] = i * i
lt += range(11, 16)
if __name__ == "__main__":
manager = Manager()
dt = manager.dict()
lt = manager.list()
p = Process(target=func, args=(dt, lt))
p.start()
p.join()
print dt, '\n', lt
执行结果：
{‘arg8': 64, ‘arg9': 81, ‘arg0': 0, ‘arg1': 1, ‘arg2': 4, ‘arg3': 9, ‘arg4': 16, ‘arg5': 25, ‘arg6': 36, ‘arg7': 49} [11, 12, 13, 14, 15]
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
