關(guān)于Python中幾種隊列Queue用法區(qū)別
python中使用到的隊列模塊大致有三個:
1、from queue import Queue
此模塊適用于線程間通信,但不能用于進程間通信。
示例代碼1: 【注意:此時代碼存在錯誤?。?!】
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()當生產(chǎn)者速度沒有消費者速度快時,上述消費者代碼會提前結(jié)束,導(dǎo)致生產(chǎn)者的速度不能消費。
while True:
x = queue.get()這種寫法也存在問題,此時消費者隊列會一直監(jiān)聽生產(chǎn)者隊列是否有數(shù)據(jù),導(dǎo)致線程一直處于阻塞狀態(tài),程序會一直阻塞不會停止,嚴重浪費系統(tǒng)資源。如果使用apscheduler等定時任務(wù)的庫的話,會導(dǎo)致定時任務(wù)無法啟動。
其實queue隊列中的put()或者get()方法中都提供了timeout參數(shù),利用這個參數(shù)可以有效解決上述消除不能消費和線程一直阻塞問題。
示例代碼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é)束!")運行結(jié)果:

根據(jù)不同的情境,可以根據(jù)實際情況設(shè)置timeout的值。如果使用定時任務(wù),使用timeout是可以的,不會使程序拋異常停止的。
2、from multiprocessing import Queue
此模塊用于對進程,但是不能用于進程池
示例代碼:
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()會報錯
# 使用multiprocessing中的Queue(),正確輸出a
運行結(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中的進程間通信需要使用Manager
pool.apply_async(producer, args=(queue, ))
pool.apply_async(consumer, args=(queue, ))
pool.close()
pool.join()運行結(jié)果:

到此這篇關(guān)于關(guān)于Python中幾種隊列Queue用法區(qū)別的文章就介紹到這了,更多相關(guān)Python中的隊列Queue內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Biblibili視頻投稿接口分析并以Python實現(xiàn)自動投稿功能
這篇文章主要介紹了Biblibili視頻投稿接口分析并以Python實現(xiàn)自動投稿功能,本文給大家介紹的非常詳細,對大家的學(xué)習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2021-02-02

