1 消息队列
1.1 基本语法
消息队列：multiprocessing.Queue，Queue是对进程安全的队列，可以使用Queue实现对进程之间的数据传输；还有一个重要作用是作为缓存使用。
Queue(maxsize = 0) method of multiprocessing， returns a queue obiect
Queue(maxzize = 0)创建一个队列对象，maxsize 表示队列中最多存放消息的数量。
返回一个队列对象
1.1 队列对象操作方法：
1.1.1 put方法
put(obj [, block=True[, timeout]])
调用队列对象的put()方法将obj插入到队列中，
第一个obj为必需参数，为插入项目的值；存入消息的种类不限制。
第二个block为可选参数，默认为True，
当block为True，timeout为空时，q.put([1,2,3])、q.put([1,2,3],True) 表示将序列插入到队尾，阻塞调用，如果q队列满时，一直等待（无超时限制的阻塞调用）。
当block为True，timeout为正整数时，q.put([1,],True,2) 表示阻塞调用进程最多timeout秒，如果超过该时间仍无空间可用，则抛出Queue.Full异常（带超时的阻塞调用）。
当block为False，q.put([1,], False) 表示调用进程时如果有空闲空间则将数据放入队列中，否则立即抛出Queue.Full异常。
简而言之，timeout表示超时等待时间，当队列满时，再存入消息就会发生阻塞（True条件下有效），阻塞时间超过timeout等待时间则抛出异常。
1.1.2 get方法
get([block=True[, timeout]])
get方法可以将队列中读取并删除一个元素。
实际上，get()方法的使用与put()函数类似
第一个block可选参数，默认为True。
当block为True，timeout为空时，阻塞等待取值，直到取到为止。
当block为True，timeout为正整数时，在timeout时间内没有取到任何元素，则会抛出Queue.Empty异常；
当block为False时，如果可以取到至时，则会立刻返回该值，如果没有取到元素则会立即抛出Queue.Empty异常。
1.1.3 full()
q.full() 判断队列是否为满；若满，返回True，若不满，返回False
1.1.4 empty()
q.empty() 判断队列是否为空，若为空，则返回True
1.1.5 qsize()
q.qsize() 获取队列中消息数量
2 基础示例
示例1
from multiprocessing import Queue
#创建队列对象,队列中最大消息数量为3
q = Queue(maxsize = 3)
#存入消息
q.put(1)
q.put('hello')#以字符串存入队列中
q.put([1,2,3,4])
q.put('hello world',True,5)
运行
Traceback (most recent call last):
File "a.py", line 10, in <module>
q.put('hello world',True,5)
File "/usr/lib/python3.5/multiprocessing/queues.py", line 83, in put
raise Full
queue.Full
因最后存入时已经超过队列的最大数量，阻塞5秒后报错
示例2
from multiprocessing import Queue
#创建队列对象,队列中最大消息数量为3
q = Queue(maxsize = 3)
#存入消息
q.put(1)
q.put('hello')#以字符串存入队列中
q.put([1,2,3,4])
# 判断队列是否为满
print(q.full())# True
# 队列中消息数量
# 判断队列是否为空
print(q.empty()) # False
# 取出消息
print(q.get()) # hello
print(q.get()) # [1,2,3,4]
#判断队列是否为空
print(q.empty()) #True
print(q.get(True,3)) #报错queue.Empty
运行
True
3
False
1
2
hello
[1, 2, 3, 4]
True
Traceback (most recent call last):
File "a.py", line 25, in <module>
print(q.get(True,3)) #报错
File "/usr/lib/python3.5/multiprocessing/queues.py", line 105, in get
raise Empty
queue.Empty
3 综合应用
多个进程间的通信 示例1
import multiprocessing
def writer_proc(q):
try:
q.put(1, block = False)
except:
pass
def reader_proc(q):
try:
print(q.get(block = False))
except:
pass
if __name__ == "__main__":
q = multiprocessing.Queue()
writer = multiprocessing.Process(target=writer_proc, args=(q,))
writer.start()
reader = multiprocessing.Process(target=reader_proc, args=(q,))
reader.start()
reader.join()
writer.join()
运行结果为 1 或 在终端上没有任何显示
实际上，这时由于多线程代码各自运行所致，如果取出先运行，同时又有报错处理设置，所以不会报错；而是没有任何输出。
为了达到目的，可以在取出函数中增加sleep()函数。
import multiprocessing
import time
def writer_proc(q):
try:
q.put(1, block = False)
except:
pass
def reader_proc(q):
time.sleep(1)
try:
print(q.get(block = False))
except:
pass
if __name__ == "__main__":
q = multiprocessing.Queue()
writer = multiprocessing.Process(target=writer_proc, args=(q,))
writer.start()
reader = multiprocessing.Process(target=reader_proc, args=(q,))
reader.start()
reader.join()
writer.join()
经过改良后，每一次运行代码均可达到输入输出的目的。
多个子进程间的通信示例2
多个子进程间的通信就要采用第一步中的队列Queue，比如，有以下需求，一个子进程向队列中写数据，另一个进程从队列中取数据，
# _*_ encoding:utf-8 _*_
from multiprocessing import Process,Queue,Pool,Pipe
import os,time,random
#写数据进程执行的代码：
def write(p):
for value in ['A','B','C']:
print ('Write---Before Put value---Put %s to queue...' % value)
p.put(value)
print ('Write---After Put value')
time.sleep(random.random())
print ('Write---After sleep')
#读数据进程执行的代码：
def read(p):
while True:
print ('Read---Before get value')
value = p.get(True)
print ('Read---After get value---Get %s from queue.' % value)
if __name__ == '__main__':
#父进程创建Queue，并传给各个子进程：
p = Queue()
pw = Process(target=write,args=(p,))
pr = Process(target=read,args=(p,))
#启动子进程pw，写入：
pw.start()
#启动子进程pr，读取:
pr.start()
#等待pw结束：
pw.join()
#pr进程里是死循环，无法等待其结束，只能强行终止：
pr.terminate()
运行
Write---Before Put value---Put A to queue... Read---Before get value Write---After Put value Read---After get value---Get A from queue. Read---Before get value Write---After sleep Write---Before Put value---Put B to queue... Write---After Put value Read---After get value---Get B from queue. Read---Before get value Write---After sleep Write---Before Put value---Put C to queue... Write---After Put value Read---After get value---Get C from queue. Read---Before get value Write---After sleep
关于锁的应用，在不同程序间如果有同时对同一个队列操作的时候，为了避免错误，可以在某个函数操作队列的时候给它加把锁，这样在同一个时间内则只能有一个子进程对队列进行操作，锁也要在manager对象中的锁
示例3
from multiprocessing import Process,Queue
import time
process_list = []
q = Queue()
def fun(name):
time.sleep(1)
q.put('hello' + str(name))
for i in range(10):
p = Process(target = fun,args = (i,))
p.start()
process_list.append(p)
for i in process_list:
i.join()
while not q.empty():
print(q.get())
运行
hello0 hello1 hello3 hello2 hello4 hello6 hello5 hello9 hello8 hello7
参考：