8.python多进程之队列

一、队列

• IPC（Inter-Process Communication）进程之间通信
• 队列先进先出
• Queue()实例化是可以传一个参数代表队列上限，如果不传或传负数都表示队列没有限制。
• 当队列容量不够时会阻塞
• Queue可以创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递

---

二、常用属性和方法

1. Queue([maxsize])

   • 创建共享的进程队列。
   • 参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。
   • 底层队列使用管道和锁定实现。

2. q.get( [ block [ ,timeout ] ] )

   • 返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。
   • block用于控制阻塞行为，默认为True阻塞进程. 如果设置为False，不阻塞但将引发Queue.Empty异常（定义在Queue模块中）。
   • timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。

3. q.get_nowait( )

   • 同q.get(False)方法。

4. q.put(item [, block [,timeout ] ] )

   • 将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。
   • block控制阻塞行为，默认为True阻塞。如果设置为False，不阻塞但将引发Queue.Empty异常（定义在Queue库模块中）。
   • timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。

5. q.put_nowait()

   • 同q.put(False)方法

6. q.qsize()

   • 返回队列中目前项目的正确数量。
   • 此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。
   • 在某些系统上，此方法可能引发NotImplementedError异常。

7. q.empty()

   • 如果调用此方法时 q为空，返回True。
   • 如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

8. q.full()

   • 如果q已满，返回为True.
   • 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）

9. q.close()

   • 关闭队列，防止队列中加入更多数据。
   • 调用此方法时，后台进程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。
   • 如果q被垃圾收集，将自动调用此方法。
   • 关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。

10. q.cancel_join_thread()

    • 不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。

11. q.join_thread()

    • 连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。
    • 调用q.cancel_join_thread()方法可以禁止这种行为。

---

三、 实例

例1. Queue 常用方法使用示例

from multiprocessing import Queue
q = Queue(3)

q.put(3)
q.put(3)
q.put(3)
# q.put(3)#如果队列已经满了，程序就会停在这里，等待数据别取走之后再将数据放入队列
try:
    q.put_nowait(3) #队列满了不会阻塞但会报错。
                    #因此可以用try语句来处理这个错误，这样程序不会一直阻塞下    
                    #去，但是会丢掉这个消息。
except:
    print('队列已满了')
#因此在放入数据前可以先看一下队列的状态，如果已经满了，就不继续put了
print(q.full())

print(q.get())
print(q.get())
print(q.get())
# print(q.get()) #同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。
try:
    q.get_nowait()
except:
    print('队列已经空了')
print(q.empty())

例2. 进程间通过队列通信示例

import os
import time
from multiprocessing import Queue, Process, freeze_support


def inputQ(queue):
    info = str(os.getpid()) +"(put):"+ str(time.asctime())
    queue.put(info)


def outputQ(queue):
    info = queue.get()
    print('%s%s \033[32m%s\033[0m' %(str(os.getpid()), '(get):',info))


if __name__ == '__main__':
    freeze_support()
    record1 = [] #store input process
    record2 = [] #stroe output process
    queue = Queue(3)

    # 输入进程
    for i in range(10):
        process = Process(target=inputQ, args=(queue,))
        process.start()
        record1.append(process)
    #输出进程
    for i in range(10):
        process = Process(target=outputQ, args=(queue,))
        process.start()
        record2.append(process)

    for p in record1:
        p.join()
    for p in record2:
        p.join()

输出结果

17700(get): 12124(put):Fri Dec 28 23:51:36 2018
13132(get): 12488(put):Fri Dec 28 23:51:37 2018
11512(get): 14304(put):Fri Dec 28 23:51:37 2018
15292(get): 424(put):Fri Dec 28 23:51:37 2018
15800(get): 15004(put):Fri Dec 28 23:51:37 2018
2892(get): 16188(put):Fri Dec 28 23:51:37 2018
16220(get): 16952(put):Fri Dec 28 23:51:37 2018
11016(get): 11268(put):Fri Dec 28 23:51:37 2018
13452(get): 10376(put):Fri Dec 28 23:51:38 2018
15264(get): 2128(put):Fri Dec 28 23:51:38 2018