怎么在Python中使用Queue消息隊列

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

進程間通信-Queue

Process之間有時需要通信，操作系統提供了很多機制來實現進程間的通信。

1. Queue的使用

可以使用multiprocessing模塊的Queue實現多進程之間的數據傳遞，Queue本身是一個消息列隊程序，首先用一個小實例來演示下Queue的工作原理：
代碼如下：

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

運行結果：

False
True
消息隊列已滿，現有消息數量：3
消息隊列已滿，現有消息數量：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中讀數據：

#coding=utf-8
from multiprocessing import Queue, Process
import time, random, os
#寫數據進程執行的代碼
def write(q):
  l1 = ['A','B','C']
  for value in l1:
    print('put %s to queue...'%value)
    q.put(value)
    time.sleep(random.random())
#讀數據執行的代碼
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__":
  #父進程創建Queue,并傳給各個子進程
  q = Queue()
  qw = Process(target=write, args=(q,))
  qr = Process(target=read, args=(q,))
  #啟動子進程ｑｗ寫入
  qw.start()
  qw.join()
  # 啟動子進程ｑｒ寫入
  qr.start()
  qr.join()
  # ｑｒ進程是死循環，無法等待其結束，只能強行終止:
  print('所有數據都已經寫入并讀取完畢')

運行結果：

put A to queue...
put B to queue...
put C to queue...
get A from queue...
get B from queue...
get C from queue...
所有數據都已經寫入并讀取完畢

3. 進程池中的Queue

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

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

下面的實例演示了進程池中的進程如何通信：

代碼如下：

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

運行結果：

父進程7415啟動．．．
writer啟動7421,父進程為7415
reader啟動7422，　父進程為7415
reader從Queue獲取到消息：a
reader從Queue獲取到消息：b
reader從Queue獲取到消息：c
reader從Queue獲取到消息：d
reader從Queue獲取到消息：e
7415結束

看完上述內容，你們對怎么在Python中使用Queue消息隊列有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業資訊頻道，感謝大家的支持。