Queue隊(duì)列中join()與task_done()的關(guān)系及說(shuō)明
join()與task_done()的關(guān)系
在網(wǎng)上大多關(guān)于join()與task_done()的結(jié)束原話是這樣的:
Queue.task_done()在完成一項(xiàng)工作之后,Queue.task_done()函數(shù)向任務(wù)已經(jīng)完成的隊(duì)列發(fā)送一個(gè)信號(hào)Queue.join()實(shí)際上意味著等到隊(duì)列為空,再執(zhí)行別的操作
但是可能很多人還是不太理解,這里以我自己的理解來(lái)闡述這兩者的關(guān)聯(lián)。
理解
如果線程里每從隊(duì)列里取一次,但沒(méi)有執(zhí)行task_done(),則join無(wú)法判斷隊(duì)列到底有沒(méi)有結(jié)束,在最后執(zhí)行個(gè)join()是等不到結(jié)果的,會(huì)一直掛起。
可以理解為,每task_done一次 就從隊(duì)列里刪掉一個(gè)元素,這樣在最后join的時(shí)候根據(jù)隊(duì)列長(zhǎng)度是否為零來(lái)判斷隊(duì)列是否結(jié)束,從而執(zhí)行主線程。
下面看個(gè)自己寫的例子:
下面這個(gè)例子,會(huì)在join()的地方無(wú)限掛起,因?yàn)閖oin在等隊(duì)列清空,但是由于沒(méi)有task_done,它認(rèn)為隊(duì)列還沒(méi)有清空,還在一直等。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以為什么要用線程,因?yàn)榫€程可以start后繼續(xù)執(zhí)行后面的主線程,可以put數(shù)據(jù),如果不是線程直接在get阻塞。
class Mythread(threading.Thread):
def __init__(self,que):
threading.Thread.__init__(self)
self.queue = que
def run(self):
while True:
sleep(1)
if self.queue.empty(): #判斷放到get前面,這樣可以,否則隊(duì)列最后一個(gè)取完后就空了,直接break,走不到print
break
item = self.queue.get()
print(item,'!')
#self.queue.task_done()
return
que = queue.Queue()
tasks = [Mythread(que) for x in range(1)]
for x in range(10):
que.put(x) #快速生產(chǎn)
for x in tasks:
t = Mythread(que) #把同一個(gè)隊(duì)列傳入2個(gè)線程
t.start()
que.join()
print('---success---')
如果把self.queue.task_done() 注釋去掉,就會(huì)順利執(zhí)行完主程序。
這就是“ Queue.task_done()函數(shù)向任務(wù)已經(jīng)完成的隊(duì)列發(fā)送一個(gè)信號(hào)”這句話的意義,能夠讓join()函數(shù)能判斷出隊(duì)列還剩多少,是否清空了。
而事實(shí)上我們看下queue的源碼可以看出確實(shí)是執(zhí)行一次未完成隊(duì)列減一:
def task_done(self):
'''Indicate that a formerly enqueued task is complete.
Used by Queue consumer threads. For each get() used to fetch a task,
a subsequent call to task_done() tells the queue that the processing
on the task is complete.
If a join() is currently blocking, it will resume when all items
have been processed (meaning that a task_done() call was received
for every item that had been put() into the queue).
Raises a ValueError if called more times than there were items
placed in the queue.
'''
with self.all_tasks_done:
unfinished = self.unfinished_tasks - 1
if unfinished <= 0:
if unfinished < 0:
raise ValueError('task_done() called too many times')
self.all_tasks_done.notify_all()
self.unfinished_tasks = unfinished
快速生產(chǎn)-快速消費(fèi)
上面的演示代碼是快速生產(chǎn)-慢速消費(fèi)的場(chǎng)景,我們可以直接用task_done()與join()配合,來(lái)讓empty()判斷出隊(duì)列是否已經(jīng)結(jié)束。
當(dāng)然,queue我們可以正確判斷是否已經(jīng)清空,但是線程里的get隊(duì)列是不知道,如果沒(méi)有東西告訴它,隊(duì)列空了,因此get還會(huì)繼續(xù)阻塞,那么我們就需要在get程序中加一個(gè)判斷,如果empty()成立,break退出循環(huán),否則get()還是會(huì)一直阻塞。
慢速生產(chǎn)-快速消費(fèi)
但是如果生產(chǎn)者速度與消費(fèi)者速度相當(dāng),或者生產(chǎn)速度小于消費(fèi)速度,則靠task_done()來(lái)實(shí)現(xiàn)隊(duì)列減一則不靠譜,隊(duì)列會(huì)時(shí)常處于供不應(yīng)求的狀態(tài),常為empty,所以用empty來(lái)判斷則不靠譜。
那么這種情況會(huì)導(dǎo)致 join可以判斷出隊(duì)列結(jié)束了,但是線程里不能依靠empty()來(lái)判斷線程是否可以結(jié)束。
我們可以在消費(fèi)隊(duì)列的每個(gè)線程最后塞入一個(gè)特定的“標(biāo)記”,在消費(fèi)的時(shí)候判斷,如果get到了這么一個(gè)“標(biāo)記”,則可以判定隊(duì)列結(jié)束了,因?yàn)樯a(chǎn)隊(duì)列都結(jié)束了,也不會(huì)再新增了。
代碼如下:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以為什么要用線程,因?yàn)榫€程可以start后繼續(xù)執(zhí)行后面的主線程,可以put數(shù)據(jù),如果不是線程直接在get阻塞。
class Mythread(threading.Thread):
def __init__(self,que):
threading.Thread.__init__(self)
self.queue = que
def run(self):
while True:
item = self.queue.get()
self.queue.task_done() #這里要放到判斷前,否則取最后最后一個(gè)的時(shí)候已經(jīng)為空,直接break,task_done執(zhí)行不了,join()判斷隊(duì)列一直沒(méi)結(jié)束
if item == None:
break
print(item,'!')
return
que = queue.Queue()
tasks = [Mythread(que) for x in range(1)]
#快速生產(chǎn)
for x in tasks:
t = Mythread(que) #把同一個(gè)隊(duì)列傳入2個(gè)線程
t.start()
for x in range(10):
sleep(1)
que.put(x)
for x in tasks:
que.put(None)
que.join()
print('---success---')注意點(diǎn)
put隊(duì)列完成的時(shí)候千萬(wàn)不能用task_done(),否則會(huì)報(bào)錯(cuò):
task_done() called too many times
因?yàn)樵摲椒▋H僅表示get成功后,執(zhí)行的一個(gè)標(biāo)記。
總結(jié)
以上為個(gè)人經(jīng)驗(yàn),希望能給大家一個(gè)參考,也希望大家多多支持腳本之家。
相關(guān)文章
python基礎(chǔ)之文件處理知識(shí)總結(jié)
今天帶大家了解python文件處理的相關(guān)知識(shí),文中介紹的非常詳細(xì),對(duì)正在學(xué)習(xí)python的小伙伴們很有幫助,需要的朋友可以參考下2021-05-05
Python?PDF轉(zhuǎn)化wolrd代碼的寫法小結(jié)
將PDF文件轉(zhuǎn)換為Word文檔的過(guò)程通常需要使用一些外部庫(kù)來(lái)實(shí)現(xiàn),因?yàn)镻ython本身并不直接支持這種轉(zhuǎn)換,這篇文章主要介紹了Python?PDF轉(zhuǎn)化wolrd代碼的寫法小結(jié),需要的朋友可以參考下2024-06-06
Python+selenium實(shí)現(xiàn)自動(dòng)循環(huán)扔QQ郵箱漂流瓶
這篇文章主要為大家詳細(xì)介紹了Python+selenium實(shí)現(xiàn)自動(dòng)循環(huán)扔QQ郵箱漂流瓶,文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2018-05-05
Python必備技巧之Pandas數(shù)據(jù)合并函數(shù)
Pandas中一共有五個(gè)數(shù)據(jù)合并函數(shù),分別為:concat、append、merge、join、combine,本文詳細(xì)講解這五個(gè)函數(shù)的使用方法,需要的可以參考一下2022-03-03
pycharm安裝opencv-python報(bào)錯(cuò)的解決
本文主要介紹了pycharm安裝opencv-python報(bào)錯(cuò)的解決,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-07-07

