Python多进程编程-进程间协作（Queue、Lock、Semaphore、Event、Pipe）

进程与进程之间是相互独立的，互不干扰。如果多进程之间需要对同一资源操作，就需要进程间共享变量，上一篇文章介绍了进程间共享数据的三大类Value、Array、Manager，这三种类的主要区别在于管理的数据类型不同。解决了进程间共享数据的问题，又有新的问题产生，那就是当多进程同时对一个共享资源进行写操作时，容易造成数据混乱。如何解决这一问题呢？这就需要进程间相互协同。

进程间相互协同的方式有如下几种：

Lock：加锁，accquire()申请锁，release()释放锁

Queue:队列

Semaphore：信号量

Event:事件

Pipe:管道

下面一一介绍这几种方式。

Lock：进程申请锁后，会阻塞其他进程获取lock对象，不能操作数据，只有锁释放后，其他进程可以重新获取锁对象。

实例代码：

from multiprocessing import Process,Value,Lock
def func1(num,lock):#传入的lock对象
lock.acquire()#申请锁，lock对象变为locked,并且阻塞其他进程获取lock对象
print num.value
num.value+=1
lock.release()#释放锁，lock对象变为unlocked,其他进程可以重新获取lock对象
if __name__ == '__main__':
lock=Lock()#创建锁对象
val=Value('i',0)
proc=[Process(target=func1,args=(val,lock)) for i in xrange(100)]
for p in proc:
p.start()
for p in proc:
p.join()

print val.value

运行结果：

50

Semaphore：信号量，同时允许多个进程申请锁

实例代码：

import multiprocessing
import time
from multiprocessing import Process,Semaphore
def func1(s):
s.acquire()
print multiprocessing.current_process().name+'acquire'
time.sleep(10)
s.release()
print multiprocessing.current_process().name+'release'
if __name__ == '__main__':
s=Semaphore(4)
procs=[Process(target=func1,args=(s,))for i in xrange(10)]
for p in procs:
p.start()
for p in procs:
p.join()

Queue:队列：多用来在多个进程间通信

实例代码：

循环等待，避免执行结果不准*

import multiprocessing
import time
def write(q):
for value in ['A','B','C']:
print 'put %s to queue '%value
q.put(value)
time.sleep(2)
def read(q):
def isEmpty():
i=0
while i<5:
if not q.empty():
return False#如果队列不等于空，退出循环不用等待
time.sleep(2)#如果列表是空，循环等待
i+=1
return True
while not isEmpty():#q.empty():
print 'get %s from queue '%q.get()
if __name__ == '__main__':
q=multiprocessing.Queue()
pw=multiprocessing.Process(target=write,args=(q,))
pr=multiprocessing.Process(target=read,args=(q,))
pw.start()
pr.start()

运行结果：

put A to queue
get A from queue
put B to queue
get B from queue
put C to queue
get C from queue

不用循环等待的效果执行结果：

put A to queue
get A from queue
put B to queue
put C to queue

Event：处理某个进程的逻辑，依赖于某个事件的发生。比如到下课时间响铃，那么这个事件就可以设置成时间，来监听到没到下课时间，如果到，代码分支就是放一段音乐。

set()设置事件，wait([time]) clear()清除事件

实例代码：

import multiprocessing
import time
def waitForEvent(e):
print 'wait for event starting.No timeout'
e.wait()
print 'the event has happened.',e.is_set()
def waitForEventTimeout(e,t):
print 'wait for event starting.timeout is ',t
e.wait(t)
print 'the event has happened.',e.is_set()
if __name__ == '__main__':
e=multiprocessing.Event()
print 'the event is_set is:',e.is_set()
p=multiprocessing.Process(target=waitForEventTimeout,args=(e,2))#改成5，True
# p1=multiprocessing.Process(target=waitForEvent,args=(e,))
p.start()
# p1.start()
time.sleep(3)
e.set()

运行结果：

the event is_set is: False
wait for event starting.timeout is  2
the event has happened. False

Pipe：管道。用来在两个进程间进行通信。

返回两个连接对象（conn1,conn2）,代表管道的两端。

send() recv()

实例代码：

from multiprocessing import Process,Pipe
def func1(pipe):
pipe.send('hello')
print 'func1 received: %s'%pipe.recv()#recv是阻塞式的，等待接收消息
pipe.send('who are you?')
print 'func1 received: %s'%pipe.recv()
def func2(pipe):
print 'func2 received: %s'%pipe.recv()
pipe.send('hello too')
print 'func2 received: %s'%pipe.recv()
pipe.send("I don't tell you")
if __name__ == '__main__':
pipe=Pipe()
print len(pipe)
p1=Process(target=func1,args=(pipe[0],))
p2=Process(target=func2,args=(pipe[1],))
p1.start()
p2.start()

执行结果：

2
func2 received: hello
func1 received: hello too
func2 received: who are you?
func1 received: I don't tell you

循环发送：A发一次，B回复一次收到

实例代码：

from multiprocessing import Process,Pipe
import time
def func1(pipe):
for i in xrange(10):
pipe.send('send message to func2:%s'%str(i))
recvMesg= pipe.recv()
print recvMesg
# with open('D:\\python\\pipeTestFun1.txt','w')as fp:
#     fp.write(recvMesg)
def func2(pipe):
for i in xrange(10):
recvMesg=pipe.recv()
# with open('D:\\python\\pipeTestFun2.txt','w') as fp:
#     fp.write(recvMesg+'\n')
sendMesg=recvMesg+' is writen to filePath'
print recvMesg
pipe.send(sendMesg)
if __name__ == '__main__':
pipe=Pipe()
p1=Process(target=func1,args=(pipe[0],))
p2=Process(target=func2,args=(pipe[1],))
p2.start()
time.sleep(1)
p1.start()

执行结果：

send message to func2:0
send message to func2:0 is writen to filePath
send message to func2:1
send message to func2:1 is writen to filePath
send message to func2:2
send message to func2:2 is writen to filePath
send message to func2:3
send message to func2:3 is writen to filePath
send message to func2:4
send message to func2:4 is writen to filePath