怎么在Python中使用Queue消息隊(duì)列

今天就跟大家聊聊有關(guān)怎么在Python中使用Queue消息隊(duì)列，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

進(jìn)程間通信-Queue

Process之間有時(shí)需要通信，操作系統(tǒng)提供了很多機(jī)制來(lái)實(shí)現(xiàn)進(jìn)程間的通信。

1. Queue的使用

可以使用multiprocessing模塊的Queue實(shí)現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞，Queue本身是一個(gè)消息列隊(duì)程序，首先用一個(gè)小實(shí)例來(lái)演示下Queue的工作原理：
代碼如下：

#coding=utf-8
from multiprocessing import Queue
#初始化一個(gè)Queue對(duì)象，最多可接收三條put消息
q = Queue(3)
q.put('消息１')
q.put('消息２')
print(q.full())#False
q.put('消息３')
print(q.full())#True
#因?yàn)橄⒘嘘?duì)已滿下面的try都會(huì)拋出異常，第一個(gè)try會(huì)等待2秒后再拋出異常，第二個(gè)Try會(huì)立刻拋出異常
try:
  q.put('消息４', True, 2)
except:
  print('消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：%s'%q.qsize())
try:
  q.put_nowait('消息４')#等同于q.put('消息４', False)
except:
  print('消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：%s'%q.qsize())
#推薦的方式，先判斷消息列隊(duì)是否已滿，再寫(xiě)入
if not q.full():
  q.put_nowait('消息４')
#讀取消息時(shí)，先判斷消息列隊(duì)是否為空，再讀取
if not q.empty():
  for i in range(q.qsize()):
    print(q.get_nowait())

運(yùn)行結(jié)果：

False
True
消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：3
消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：3
消息１
消息２
消息３

說(shuō)明

初始化Queue()對(duì)象時(shí)（例如：q=Queue()），若括號(hào)中沒(méi)有指定最大可接收的消息數(shù)量，或數(shù)量為負(fù)值，那么就代表可接受的消息數(shù)量沒(méi)有上限（直到內(nèi)存的盡頭）；

Queue.qsize()：返回當(dāng)前隊(duì)列包含的消息數(shù)量；
Queue.empty()：如果隊(duì)列為空，返回True，反之False ；
Queue.full()：如果隊(duì)列滿了，返回True,反之False；
Queue.get([block[, timeout]])：獲取隊(duì)列中的一條消息，然后將其從列隊(duì)中移除，block默認(rèn)值為T(mén)rue；

1）如果block使用默認(rèn)值，且沒(méi)有設(shè)置timeout（單位秒），消息列隊(duì)如果為空，此時(shí)程序?qū)⒈蛔枞ㄍＴ谧x取狀態(tài)），直到從消息列隊(duì)讀到消息為止，如果設(shè)置了timeout，則會(huì)等待timeout秒，若還沒(méi)讀取到任何消息，則拋出”Queue.Empty”異常；

2）如果block值為False，消息列隊(duì)如果為空，則會(huì)立刻拋出”Queue.Empty”異常；
Queue.get_nowait()：相當(dāng)Queue.get(False)；
Queue.put(item,[block[, timeout]])：將item消息寫(xiě)入隊(duì)列，block默認(rèn)值為T(mén)rue；

1）如果block使用默認(rèn)值，且沒(méi)有設(shè)置timeout（單位秒），消息列隊(duì)如果已經(jīng)沒(méi)有空間可寫(xiě)入，此時(shí)程序?qū)⒈蛔枞ㄍＴ趯?xiě)?狀態(tài)），直到從消息列隊(duì)騰出空間為止。如果設(shè)置了timeout，則會(huì)等待timeout秒，若還沒(méi)空間，則拋出”Queue.Full”異常；

2）如果block值為False，消息列隊(duì)如果沒(méi)有空間可寫(xiě)入則會(huì)立刻拋出”Queue.Full”異常；

Queue.put_nowait(item)：相當(dāng)Queue.put(item, False)；

2. Queue實(shí)例

我們以Queue為例，在子進(jìn)程中創(chuàng)建兩個(gè)子進(jìn)程，一個(gè)往Queue中寫(xiě)數(shù)據(jù)，一個(gè)從Queue中讀數(shù)據(jù)：

#coding=utf-8
from multiprocessing import Queue, Process
import time, random, os
#寫(xiě)數(shù)據(jù)進(jìn)程執(zhí)行的代碼
def write(q):
  l1 = ['A','B','C']
  for value in l1:
    print('put %s to queue...'%value)
    q.put(value)
    time.sleep(random.random())
#讀數(shù)據(jù)執(zhí)行的代碼
def read(q):
  while True:
    if not q.empty():
      value = q.get(True)
      print('get %s from queue...' % value)
      time.sleep(random.random())
    else:
      break
if __name__ == "__main__":
  #父進(jìn)程創(chuàng)建Queue,并傳給各個(gè)子進(jìn)程
  q = Queue()
  qw = Process(target=write, args=(q,))
  qr = Process(target=read, args=(q,))
  #啟動(dòng)子進(jìn)程ｑｗ寫(xiě)入
  qw.start()
  qw.join()
  # 啟動(dòng)子進(jìn)程ｑｒ?qū)懭?
  qr.start()
  qr.join()
  # ｑｒ進(jìn)程是死循環(huán)，無(wú)法等待其結(jié)束，只能強(qiáng)行終止:
  print('所有數(shù)據(jù)都已經(jīng)寫(xiě)入并讀取完畢')

運(yùn)行結(jié)果：

put A to queue...
put B to queue...
put C to queue...
get A from queue...
get B from queue...
get C from queue...
所有數(shù)據(jù)都已經(jīng)寫(xiě)入并讀取完畢

3. 進(jìn)程池中的Queue

如果要使用Pool創(chuàng)建進(jìn)程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否則會(huì)得到一條如下的錯(cuò)誤信息：

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

下面的實(shí)例演示了進(jìn)程池中的進(jìn)程如何通信：

代碼如下：

#coding=utf-8
from multiprocessing import Manager, Pool
import time, random, os
def writer(q):
  print('writer啟動(dòng)%s,父進(jìn)程為%s'%(os.getpid(),os.getppid()))
  l1 = ['a','b','c','d','e']
  for value in l1:
    q.put(value)
def reader(q):
  print('reader啟動(dòng)%s，父進(jìn)程為%s'%(os.getpid(),os.getppid()))
  for i in range(q.qsize()):
    print('reader從Queue獲取到消息：%s'%q.get(True))
if __name__ == "__main__":
  print('父進(jìn)程%s啟動(dòng)．．．'%os.getpid())
  q = Manager().Queue() #使用Manager中的Queue來(lái)初始化
  po = Pool()
  # 使用阻塞模式創(chuàng)建進(jìn)程，這樣就不需要在reader中使用死循環(huán)了，可以讓writer完全執(zhí)行完成后，再用reader去讀取
  po.apply(writer, (q,))
  po.apply(reader, (q,))
  po.close()
  po.join()
  print('%s結(jié)束'%os.getpid())

運(yùn)行結(jié)果：

父進(jìn)程7415啟動(dòng)．．．
writer啟動(dòng)7421,父進(jìn)程為7415
reader啟動(dòng)7422，　父進(jìn)程為7415
reader從Queue獲取到消息：a
reader從Queue獲取到消息：b
reader從Queue獲取到消息：c
reader從Queue獲取到消息：d
reader從Queue獲取到消息：e
7415結(jié)束

看完上述內(nèi)容，你們對(duì)怎么在Python中使用Queue消息隊(duì)列有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請(qǐng)關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。