进程间的通信-Queue

1. 使用Queue

Queue可以使用multiprocesing模块来实现多过程之间的数据传输。Queue本身就是一个消息排队程序。首先，用一个小例子来演示Queue的工作原理：

#-*-coding:utf-8-*-
frommultiprocessingimportQueue
#创建一个Queue对象最多可以接受三条put消息
q=Queue(3)
q.put("消息1")
q.put("消息2")
print(q.full())
q.put("消息3")
print(q.full())
try:
q.put("消息4",True,2)
except:
print("消息队列已满，现有消息数量为%s"%q.qsize())
try:
q.put_nowait("消息5")
except:
print("消息队列已满，现有消息数量为%s"%q.qsize())
#推荐方法，首先判断消息队列是否已满，并将其写入
ifnotq.full():
q.put_nowait("消息6")
#读取消息时，首先要判断消息队列是否为空，在读取
ifnotq.empty():
foriinrange(q.qsize()):
print(q.get_nowait())

运行结果如下：

False
True
消息队列已满，现有消息数量：3
消息队列已满，现有消息数量：3
消息1
消息2
消息3

说明

在初始化Queue()对象(例如:q=Queue()，如果括号中没有指定可接收信息的数量，或者数量为负值，则表示可接收信息的数量没有上限(直到内存结束)；

Queue.qsize()：返回当前队列中包含的信息数量；

Queue.empty()：如果队列是空的，返回True，相反，False ；

Queue.full()：如果队列满了，回到True，相反，False；

Queue.get([block[, timeout]])：获取队列中的一条信息，然后从队列中删除，block默认值为True；

1)如果block使用默认值，且没有设置timeout(单位秒)，如果消息列队为空，此时程序将被堵塞(停止读取状态)，直到从消息列队读取消息。如果设置timeout，将等待timeout秒。如果没有读到任何消息，就扔掉。"Queue.Empty"异常；

2)如果block值为False，如果消息列队为空，将立即抛出"Queue.Empty"异常；

Queue.get_nowait()：相当Queue.get(False)；

Queue.put(item,[block[, timeout]])：将item消息写入队列，block默认值为true；

1)如果block使用默认值，且没有timeout(单位秒)，如果没有写信的空间，此时程序将被堵塞(停止写信)，直到消息排队腾出空间。如果设置timeout，将等待timeout秒。如果没有空间，就扔掉。"Queue.Full"异常；

2)如果block值为false，如果没有空间可以写入消息列队，则会立即抛出"Queue.Full"异常；

Queue.put_nowait(item)：相当Queue.put(item, False)；

2. Queue实例

以Queue为例，在父亲的过程中创建了两个子过程，一个在Queue上写数据，另一个在Queue上读数据：

frommultiprocessingimportProcess
frommultiprocessingimportQueue
importos
importtime
importrandom
#编写数据流程执行的代码
defwrite(q):
forvaluein["A","B","C"]:
print("Put%stoQueue"%value)
q.put(value)
time.sleep(random.random())

#读取数据过程的代码
defread(q):
whileTrue:
ifnotq.empty():
value=q.get(True)
print("Get%sfromQueue"%value)
time.sleep(random.random())
else:
break
if__name__='__main__':
#父亲创建Queue进程，并将其传递给每个子过程
q=Queue()
pw=Process(target=write,args=(q,))
pr=Process(target=read,args=(q,))
#启动子进程pw，写入
pw.start()
#等待pw结束
pw.join()
#启动子进程pr，读取
pr.start()
pr.join()
print("所有数据都被写入并阅读")

运行结果如下：

PutAtoQueue
PutBtoQueue
PutCtoQueue
GetAfromQueue
GetBfromQueue
GetCfromQueue

所有数据都写入并阅读。

3. Queueuee进程池

要使用Pool创建过程，需要使用multiprocessing.Manager()中的Queue()，而不是multiprocesing.Queue()，否则，您将获得以下错误信息：

RuntimeError: Queue objects should only be shared between processes through inheritance.

#coding=utf-8
frommultiprocessingimportManager
frommultiprocessingimportPool
importos
importtime
importrandom
defreader(q):
print("reader启动(%)d),父亲的过程(%)d)"%(os.getpid(),os.getppid()))
foriinrange(q.qsize()):
print("当reader从Queue获得消息时：%s"%q.get(True))
defwriter(q):
print("writer启动(%)d),父亲的过程是(%)d)"%(os.getpid(),os.getppid()))
foriin"Se7eN_HOU":
q.put(i)
if__name__='__main__':
print("-------(%d)Start-------"%os.getpid())
#使用Manager中的Queue初始化
q=Manager().Queue()
po=Pool()
#使用阻塞模式创建过程，这样就不需要在reader中使用死循环，可以使writer完全执行，然后用reader读取
po.apply(writer,(q,))
po.apply(reader,(q,))
po.close()
po.join()
print("-------(%d)End-------"%os.getpid())

运行结果如下：

-------(880)Start-------
writer启动(7744)，父亲的过程是(880)
reader启动(7936)，父亲的过程是(880)
当reader从Queue获得消息时：S
当reader从Queue获得消息时：e
当reader从Queue获得消息时：7
当reader从Queue获得消息时：e
当reader从Queue获得消息时：N
当reader从Queue获得消息时：_
当reader从Queue获得消息时：H
当reader从Queue获得消息时：O
当reader从Queue获得消息时：U
-------(880)End-------

相关推荐：

Python中的进程池是什么？