快捷導(dǎo)航

關(guān)于Python中幾種隊(duì)列Queue用法區(qū)別

更新時(shí)間：2023年05月01日 09:00:25 作者：IT之一小佬

這篇文章主要介紹了關(guān)于Python中幾種隊(duì)列Queue用法區(qū)別,queue隊(duì)列中的put()或者get()方法中都提供了timeout參數(shù)，利用這個(gè)參數(shù)可以有效解決上述消除不能消費(fèi)和線程一直阻塞問題,需要的朋友可以參考下

python中使用到的隊(duì)列模塊大致有三個(gè)：

1、from queue import Queue

此模塊適用于線程間通信，但不能用于進(jìn)程間通信。

示例代碼1：【注意：此時(shí)代碼存在錯(cuò)誤?。。　?/p>

import time
import threading
from queue import Queue
def task_func():
    global queue
    while queue.qsize() > 0:
        x = queue.get()
        print(f"num: {x}")
        time.sleep(0.1)
def producer_data():
    global queue
    for i in range(100):
        queue.put(i)
        time.sleep(0.1)
if __name__ == '__main__':
    queue = Queue()
    producer_thread = threading.Thread(target=producer_data)
    producer_thread.start()
    thread_list = []
    for i in range(5):
        thread = threading.Thread(target=task_func)
        thread.start()
        thread_list.append(thread)
    for thread in thread_list:
        thread.join()
    print("主程序執(zhí)行結(jié)束！")

注意：上述寫法：

    while queue.qsize() > 0:
        x = queue.get()

當(dāng)生產(chǎn)者速度沒有消費(fèi)者速度快時(shí)，上述消費(fèi)者代碼會(huì)提前結(jié)束，導(dǎo)致生產(chǎn)者的速度不能消費(fèi)。

    while True:
        x = queue.get()

這種寫法也存在問題，此時(shí)消費(fèi)者隊(duì)列會(huì)一直監(jiān)聽生產(chǎn)者隊(duì)列是否有數(shù)據(jù)，導(dǎo)致線程一直處于阻塞狀態(tài)，程序會(huì)一直阻塞不會(huì)停止，嚴(yán)重浪費(fèi)系統(tǒng)資源。如果使用apscheduler等定時(shí)任務(wù)的庫的話，會(huì)導(dǎo)致定時(shí)任務(wù)無法啟動(dòng)。

其實(shí)queue隊(duì)列中的put()或者get()方法中都提供了timeout參數(shù)，利用這個(gè)參數(shù)可以有效解決上述消除不能消費(fèi)和線程一直阻塞問題。

示例代碼2：

import time
import threading
from queue import Queue
def task_func():
    global queue
    while True:
        x = queue.get(timeout=10)
        print(f"num: {x}")
def producer_data():
    global queue
    for i in range(100):
        queue.put(i)
        time.sleep(0.1)
if __name__ == '__main__':
    queue = Queue()
    producer_thread = threading.Thread(target=producer_data)
    producer_thread.start()
    thread_list = []
    for i in range(5):
        thread = threading.Thread(target=task_func)
        thread.start()
        thread_list.append(thread)
    for thread in thread_list:
        thread.join()
    print("主程序執(zhí)行結(jié)束！")

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

根據(jù)不同的情境，可以根據(jù)實(shí)際情況設(shè)置timeout的值。如果使用定時(shí)任務(wù)，使用timeout是可以的，不會(huì)使程序拋異常停止的。

2、from multiprocessing import Queue

此模塊用于對進(jìn)程，但是不能用于進(jìn)程池

示例代碼：

import time
from multiprocessing import Process, Queue
import queue
def producer(queue):
    queue.put("a")
    time.sleep(2)
def consumer(queue):
    time.sleep(2)
    data = queue.get()
    print(data)
if __name__ == "__main__":
    # queue = queue.Queue()
    queue = Queue()
    my_producer = Process(target=producer, args=(queue, ))
    my_consumer = Process(target=consumer, args=(queue, ))
    my_producer.start()
    my_consumer.start()
    my_producer.join()
    my_consumer.join()
# 使用queue模塊的Queue()會(huì)報(bào)錯(cuò)
# 使用multiprocessing中的Queue(),正確輸出a

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

3、from multiprocessing import Manager

示例代碼：

import time
from multiprocessing import Process, Queue, Pool, Manager
def producer(queue):
    queue.put("a")
    time.sleep(2)
def consumer(queue):
    time.sleep(2)
    data = queue.get()
    print(data)
if __name__ == "__main__":
    # queue = Queue()
    queue = Manager().Queue()
    pool = Pool()
    # pool中的進(jìn)程間通信需要使用Manager
    pool.apply_async(producer, args=(queue, ))
    pool.apply_async(consumer, args=(queue, ))
    pool.close()
    pool.join()

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