Queue隊列中join()與task_done()的關(guān)系及說明

更新時間：2023年02月25日 09:52:49 作者：Terry_dong

這篇文章主要介紹了Queue隊列中join()與task_done()的關(guān)系及說明，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

join()與task_done()的關(guān)系

在網(wǎng)上大多關(guān)于join()與task_done()的結(jié)束原話是這樣的：

Queue.task_done() 在完成一項工作之后，Queue.task_done()函數(shù)向任務(wù)已經(jīng)完成的隊列發(fā)送一個信號
Queue.join() 實際上意味著等到隊列為空，再執(zhí)行別的操作

但是可能很多人還是不太理解，這里以我自己的理解來闡述這兩者的關(guān)聯(lián)。

理解

如果線程里每從隊列里取一次，但沒有執(zhí)行task_done()，則join無法判斷隊列到底有沒有結(jié)束，在最后執(zhí)行個join()是等不到結(jié)果的，會一直掛起。

可以理解為，每task_done一次就從隊列里刪掉一個元素，這樣在最后join的時候根據(jù)隊列長度是否為零來判斷隊列是否結(jié)束，從而執(zhí)行主線程。

下面看個自己寫的例子：

下面這個例子，會在join()的地方無限掛起，因為join在等隊列清空，但是由于沒有task_done，它認為隊列還沒有清空，還在一直等。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以為什么要用線程，因為線程可以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前面，這樣可以，否則隊列最后一個取完后就空了，直接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) #把同一個隊列傳入2個線程
 t.start()
 
que.join()
 
print('---success---')

如果把self.queue.task_done() 注釋去掉，就會順利執(zhí)行完主程序。

這就是“ Queue.task_done()函數(shù)向任務(wù)已經(jīng)完成的隊列發(fā)送一個信號”這句話的意義，能夠讓join()函數(shù)能判斷出隊列還剩多少，是否清空了。

而事實上我們看下queue的源碼可以看出確實是執(zhí)行一次未完成隊列減一：

 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)-快速消費

上面的演示代碼是快速生產(chǎn)-慢速消費的場景，我們可以直接用task_done()與join()配合，來讓empty()判斷出隊列是否已經(jīng)結(jié)束。

當然，queue我們可以正確判斷是否已經(jīng)清空，但是線程里的get隊列是不知道，如果沒有東西告訴它，隊列空了，因此get還會繼續(xù)阻塞，那么我們就需要在get程序中加一個判斷，如果empty()成立，break退出循環(huán)，否則get()還是會一直阻塞。

慢速生產(chǎn)-快速消費

但是如果生產(chǎn)者速度與消費者速度相當，或者生產(chǎn)速度小于消費速度，則靠task_done()來實現(xiàn)隊列減一則不靠譜，隊列會時常處于供不應(yīng)求的狀態(tài)，常為empty，所以用empty來判斷則不靠譜。

那么這種情況會導(dǎo)致 join可以判斷出隊列結(jié)束了，但是線程里不能依靠empty()來判斷線程是否可以結(jié)束。

我們可以在消費隊列的每個線程最后塞入一個特定的“標記”，在消費的時候判斷，如果get到了這么一個“標記”，則可以判定隊列結(jié)束了，因為生產(chǎn)隊列都結(jié)束了，也不會再新增了。

代碼如下：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以為什么要用線程，因為線程可以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() #這里要放到判斷前，否則取最后最后一個的時候已經(jīng)為空，直接break，task_done執(zhí)行不了，join()判斷隊列一直沒結(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) #把同一個隊列傳入2個線程
 t.start()
for x in range(10):
 sleep(1)
 que.put(x)
for x in tasks:
 que.put(None)
que.join()
print('---success---')