python—多进程的消息队列
消息队列
消息队列是在消息的传输过程中保存消息的容器
消息队列最经典的用法就是消费者 和生产者之间通过消息管道传递消息，消费者和生成者是不同的进程。生产者往管道写消息，消费者从管道中读消息
操作系统提供了很多机制来实现进程间的通信，multiprocessing模块提供了Queue和Pipe两种方法来实现
一、使用multiprocessing里面的Queue来实现消息队列
q = Queue()
q.put(data) #生产消息
data = q.get() #消费消息
例子：
from multiprocessing import Queue, Process
def write(q):
for i in ["a","b","c","d"]:
q.put(i)
print("put {0} to queue".format(i))
def read(q):
while 1:
result = q.get()
print("get {0} from queue".format(result))
def main():
q = Queue() #定义一个消息队列容器
pw = Process(target=write,args=(q,)) #定义一个写的进程
pr = Process(target=read,args=(q,)) #定义一个读的进程
pw.start() #启动进程
pr.start()
pw.join()
pr.terminate()
if __name__ == "__main__":
main()运行结果：
put a to queue
put b to queueget a from queue
get b from queue
put c to queue
put d to queue
get c from queue
get d from queue
二、通过Multiprocessing里面的Pipe来实现消息队列
1）Pipe方法返回（conn1,conn2）代表一个管道的两个端。Pipe方法有duplex参数，如果duplux参数为True(默认值)，那么这个管道是全双工模式，即conn1和conn2均可收发。duplux为False，conn1负责接收消息，conn2负责发行消息
2）send和recv方法分别是发送和接收消息的方法。close方法表示关闭管道，当消息接收结束以后，关闭管道。
例子：
from multiprocessing import Process,Pipe
import time
def proc1(pipe):
for i in xrange(1,10):
pipe.send(i)
time.sleep(3)
print("send {0} to pipe".format(i))
def proc2(pipe):
n = 9
while n>0:
result = pipe.recv()
time.sleep(3)
print("recv {0} from pipe".format(result))
n -= 1
if __name__ == "__main__":
pipe = Pipe(duplex=False) #定义并实例化一个管道
print(type(pipe))
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()
pipe[0].close()
pipe[1].close()运行结果：
<type 'tuple'>
send 1 to pipe
recv 1 from pipe
send 2 to pipe
recv 2 from pipe
recv 3 from pipe
send 3 to pipe
send 4 to piperecv 4 from pipe
send 5 to pipe
recv 5 from pipe
recv 6 from pipe
send 6 to pipe
send 7 to pipe
recv 7 from pipe
send 8 to pipe
recv 8 from pipe
recv 9 from pipesend 9 to pipe
三、Queue模块
python提供了Queue模块来专门实现消息队列：
Queue对象实现一个fifo队列（其他的还有lifo、priority队列）。queue只有gsize一个构造函数，用来指定队列容量，指定为0的时候代表容量无限。主要有以下成员函数:
Queue.gsize()：返回消息队列的当前空间。返回的值不一定可靠。
Queue.empty()：判断消息队列是否为空，返回True或者False。同样不可靠
Queue.full()：判断消息是否满
Queue.put（item,block=True,timeout=None）：往消息队列中存放数据。block可以控制是否阻塞，timeout控制阻塞时候的等待时间。如果不阻塞或者超时，会引起一个full exception。
Queue.put_nowait(item)：相当于put(item,False)
Queue.get(block=True,timeout=None)：获取一个消息，其他等同put
以下两个函数用来判断消息对应的任务是否完成：
Queue.task_done()：接收消息的线程通过调用这个函来说明消息对应的任务已完成
Queue.join()：实际上意味着等到队列为空，再执行别的操作
例子：
from multiprocessing import Process, Pipe, Queue
import time
from threading import Thread
class Proceduer(Thread):
def __init__(self,queue):
super(Proceduer,self).__init__() # 超类
self.queue = queue #将queue赋给self.queue，便于类中其他函数调用
def run(self):
try:
for i in xrange(1,10):
print("put data is: {0} to queue".format(i))
self.queue.put(i)
except Exception as e:
print("put data error")
raise e
class Consumer_odd(Thread):
def __init__(self,queue):
super(Consumer_odd, self).__init__()
self.queue = queue
def run(self):
try:
while self.queue.empty: #判断消息队列是否为空
number = self.queue.get() #取到消息值
if number%2 != 0:
print("get {0} from queue ODD".format(number))
else:
self.queue.put(number) #将信息放回队列中
time.sleep(1)
except Exception as e:
raise e
class Consumer_even(Thread):
def __init__(self,queue):
super(Consumer_even,self).__init__()
self.queue = queue
def run(self):
try:
while self.queue.empty:
number = self.queue.get()
if number%2 == 0:
print("get {0} from queue Even,thread name is :{1}".format(number,self.getName()))
else:
self.queue.put(number)
time.sleep(1)
except Exception as e:
raise e
def main():
queue = Queue() #实例化一个消息队列
p = Proceduer(queue=queue) #消息队列作为参数赋值给生产者函数，并实例化
p.start() #启动一个带消息队列的函数
p.join() #等待结束
time.sleep(1)
c1 = Consumer_odd(queue=queue) #消息队列作为参数赋值给消费者函数，并实例化
c2 = Consumer_even(queue=queue) #消息队列作为参数赋值给消费者函数，并实例化
c1.start()
c2.start()
c1.join()
c2.join()
print("All threads terminate!")
if __name__ == "__main__":
main()运行结果：
put data is: 1 to queue
put data is: 2 to queue
put data is: 3 to queue
put data is: 4 to queue
put data is: 5 to queue
put data is: 6 to queue
put data is: 7 to queue
put data is: 8 to queue
put data is: 9 to queue
get 1 from queue ODD
get 3 from queue ODD
get 4 from queue Even,thread name is :Thread-3
get 5 from queue ODD
get 7 from queue ODD
get 9 from queue ODD
get 2 from queue Even,thread name is :Thread-3
get 6 from queue Even,thread name is :Thread-3
get 8 from queue Even,thread name is :Thread-3
例子2：
import Queue
q = Queue.Queue()
for i in range(5):
q.put(i)
while not q.empty():
print q.get()运行结果：
0
1
2
3
4