一、Queue模塊的知識點思維導圖

二、Queue模塊常用函數(shù)介紹

queue模塊是內置的，不需要安裝直接導入就可以了。

（1）創(chuàng)建一個Queue對象

import queue# 創(chuàng)建一個隊列實例
q = queue.Queue(maxsize=20)   # 可選參數(shù)，默認為無限大

（2）入隊

# 導入Python標準庫中的queue模塊，該模塊提供了線程安全的隊列數(shù)據結構
import queue# 創(chuàng)建一個Queue對象，可以理解為初始化一個隊列，參數(shù)maxsize默認為0，表示隊列大小無限制
q = queue.Queue()# 使用for循環(huán)將數(shù)字0到9依次放入隊列中
# put方法用于將元素放入隊列尾部，即執(zhí)行入隊操作
for i in range(10):q.put(i)  # 這里每執(zhí)行一次，就將i（0-9）作為一個元素放入隊列中# 此時，隊列q中按照先進先出（FIFO）原則存儲了數(shù)字0-9

（3）出隊

# 導入Python內置的queue模塊，其中包含線程安全的隊列數(shù)據結構
import queue# 創(chuàng)建一個Queue對象，即初始化一個空隊列
q = queue.Queue()# 使用for循環(huán)將0到9這10個整數(shù)依次入隊
# put方法將元素添加到隊列的末尾
for i in range(10):q.put(i)  # 每次循環(huán)都將i作為元素放入隊列中

（4）判空

# 導入Python內置的queue模塊，其中包含線程安全的隊列數(shù)據結構
import queue# 創(chuàng)建一個Queue對象，即初始化一個空隊列
q = queue.Queue()# 使用for循環(huán)將0到9這10個整數(shù)依次入隊
# put方法將元素添加到隊列的末尾
for i in range(10):q.put(i)  # 每次循環(huán)都將i作為元素放入隊列中# 使用while循環(huán)檢查隊列是否為空
# 當隊列不為空時，持續(xù)執(zhí)行循環(huán)體內的操作
while not q.empty():  # 使用get方法從隊列中取出并移除一個元素# 默認情況下，get方法會阻塞直到有元素可用（除非設置了非阻塞模式）data = q.get()  # 打印從隊列中取出的元素print(data)# 當隊列為空時，while循環(huán)結束
# 此時所有之前入隊的整數(shù)0-9會被按照先進先出（FIFO）的原則依次出隊并打印

（5）判滿

import queue# 初始化一個容量為5的隊列
q = queue.Queue(maxsize=5)# 入隊操作
for i in range(10):# 在入隊前檢查隊列是否已滿if q.full():print("Queue is full, cannot insert more items.")else:q.put(i)# 若隊列已滿，可以選擇等待一段時間后重試或其他處理策略if q.full():time.sleep(1)  # 假設休眠一秒后重試continuewhile not q.empty():data = q.get()print(data)

（6）獲取隊列長度

import queue# 創(chuàng)建一個隊列
q = queue.Queue()# 將10個元素放入隊列
for i in range(10):q.put(i)# 打印隊列長度
print(f"Queue size before processing: {q.qsize()}")# 處理隊列中的元素，同時打印隊列長度
while not q.empty():data = q.get()print(f"Processing element: {data}, current queue size: {q.qsize()}")# 在處理過程中，每次從隊列中取出一個元素后，隊列長度都會減一

（7）task_done() 和 unfinished_tasks

import queue
import threading# 創(chuàng)建一個隊列
q = queue.Queue()# 生產者線程將任務放入隊列
def producer():for i in range(10):q.put(i)print(f"Producer added task: {i}")# 消費者線程從隊列中取出并處理任務
def consumer():while True:task = q.get()  # 獲取任務print(f"Consumer started working on task: {task}")# 這里模擬任務處理過程# 實際應用中，此處應替換為實際的任務處理邏輯process_task(task)# 任務處理完畢，調用 task_done()q.task_done()print(f"Consumer finished task: {task}")# 檢查隊列是否為空且所有任務都已經完成if q.empty() and q.unfinished_tasks == 0:print("All tasks completed, exiting consumer.")break# 創(chuàng)建并啟動生產者線程
p = threading.Thread(target=producer)
p.start()
p.join()  # 確保生產者線程完成任務# 創(chuàng)建并啟動消費者線程
c = threading.Thread(target=consumer)
c.start()
c.join()  # 等待消費者線程處理完所有任務并退出# 由于消費者線程在所有任務完成后退出，所以在此處不需要再次檢查隊列狀態(tài)

（8）join

import queue
import threading# 創(chuàng)建一個隊列
q = queue.Queue()# 生產者線程將任務放入隊列
def producer():for i in range(10):q.put(i)print("Producer has added all tasks.")# 消費者線程從隊列中取出并處理任務
def consumer():while True:task = q.get()  # 獲取任務print(f"Consumer started working on task: {task}")# 這里模擬任務處理過程# 實際應用中，此處應替換為實際的任務處理邏輯process_task(task)# 任務處理完畢，調用 task_done()q.task_done()print(f"Consumer finished task: {task}")# 檢查隊列是否為空if q.empty():print("No more tasks in the queue.")break# 創(chuàng)建并啟動生產者線程
p = threading.Thread(target=producer)
p.start()
p.join()  # 確保生產者線程完成任務添加# 創(chuàng)建并啟動消費者線程
c = threading.Thread(target=consumer)
c.start()# 主線程等待所有任務完成
q.join()  # 阻塞主線程，直到所有從隊列中取出的任務都被標記為完成print("All tasks have been processed and marked as done.")