python-进程间通信-队列Queue

作者：马育民 • • 阅读：10128

概述

本文讲解队列Queue实现进程间通信，即：多进程之间的数据传递

官方api网址：
https://docs.python.org/3/library/multiprocessing.html#multiprocessing.Queue

注意：
此Queue是multiprocessing模块下的类，也具有 队列 的特征，关于队列参见网址：
http://www.malaoshi.top/show_1EF2xjKH63oy.html

Queue api介绍

导入模块

from multiprocessing import Queue

创建

Queue([maxsize])

创建队列
返回使用 管道 和一些 锁 /信号灯 实现的进程共享队列。
底层是通过操作系统的 管道

参数：

  • maxsize：指定队列最多容纳元素的数量，不设置表示无穷大

返回元素数量

qsize()

返回当前队列包含的元素数量

队列是否为空

empty()

如果队列为空，返回True，反之False

队列是否满了

empty()

如果队列满了，返回True，反之False

放入元素

put(item,[block[,timeout]])

将元素放入队列

参数：

  • block：不写默认为True

    • 为True，且 没有timeout，当队列 已满，处于阻塞状态，直到有空间后，继续执行。如果设置timeout，则会等待timeout秒，还是没有空间，抛出 “queue.Full 异常”

    • 为False，如果队列 已满，会立刻抛出 “queue.Full异常”

  • timeout：超时时间，单位秒

例子：

from multiprocessing import Queue

#创建队列，最多可容纳3个元素
q=Queue(2)
#放入元素
q.put('123')
print('元素数量：%d'%q.qsize())
print('是否满了：%s'%q.full())
#放入元素
q.put([1,2,3])
print('元素数量：%d'%q.qsize())
print('是否满了：%s'%q.full())

# 释放下行代码，将一直处于阻塞状态
# q.put({'name':'张三'})
# 释放下行代码，处于阻塞状态，2秒后抛出 queue.Full异常
# q.put({'name':'张三'},True,2)
# 释放下行代码，立刻抛出 queue.Full异常
# q.put({'name':'张三'},False)

print('程序结束')

不等待放入元素

put_nowait(item)

同put(item,False)

获取元素

get([block[,timeout]])

获取队列 第一个元素，然后将其从队列中 移除

参数：

  • block：默认为True。

    • 为True，且 没有timeout，当 队列为空，处于 阻塞 状态，直到有元素，继续执行。如果 设置timeout，则会等待timeout秒，还是没有元素，抛出 “empty 异常”

    • 如果为False，如果队列为空，则会立刻抛出 “empty 异常”

  • timeout：超时时间，单位秒

例子

from multiprocessing import Queue

#创建队列，最多可容纳3个元素
q=Queue(2)
#放入元素
q.put('123')
print('元素数量：%d'%q.qsize())
print('是否满了：%s'%q.full())
#放入元素
q.put([1,2,3])
print('元素数量：%d'%q.qsize())
print('是否满了：%s'%q.full())

print('第一次获取：%s'%q.get())
print('第二次获取：%s'%q.get())

# 释放下行代码，一直处于阻塞状态
# print('第三次获取：%s'%q.get())
# 释放下行代码，处于阻塞状态，2秒后抛出queue.Empty异常
# print('第三次获取：%s'%q.get(True,2))
# 释放下行代码，立刻抛出queue.Empty异常
# print('第三次获取：%s'%q.get(False))

print('程序结束')

不等待获取元素

get_nowait()

同get(False)

例子

本例中，一个进程向队列放数据，一个进程从队列取数据

from multiprocessing import Process,Queue
import time

def putdata(qu):
    for i in range(10):
        qu.put(i)

def getdata(qu):
    while not qu.empty():
        print(qu.get())

if __name__ == "__main__":
    q=Queue()
    p=Process(target=putdata,args=(q,))
    p2=Process(target=getdata,args=(q,))
    p.start()
    p2.start()

上面这种就是 生产者消费者模式

案例

淘宝双十一秒杀时，某商品的数量假设是100件，但有101人在秒杀，100件商品就是全局变量，101人就是101个进程，在秒杀时要做判断，如果商品数量大于0就可以秒杀，否则提示售罄，编码如下：

from multiprocessing import Process,Queue
import time
import os
#双十一秒杀
def buy(qu,index):
    value=qu.get()
    print('%d号买家-开始购买,pid=%d,num=%d'%(index,os.getpid(),value))
    #修改变量
    if value>0:
        #为了便于查看结果，延迟10秒，开始购买和结束购买会分开
        # time.sleep(10)
        value-=1
        qu.put(value)
    else:
        print('售罄')
    print('%d号买家-结束购买,pid=%d,num=%d'%(index,os.getpid(),value))

if __name__ == "__main__":
    count=101
    stime=time.time()
    #秒杀商品数量
    qu=Queue()
    qu.put(100)
    ps=[]
    for index in range(count):
        p=Process(target=buy,args=(qu,index))
        ps.append(p)
    for index in range(count):
        ps[index].start()
    for index in range(count):
        ps[index].join()
    #主进程打印值
    print('主进程，num=%s'%qu.get())
    etime=time.time()
    print(etime-stime)

原文出处：/show_1EF2xlMAJUOP.html

吉ICP备15003043号