Queue隊列中join()與task_done()的關系是什么

這篇文章主要介紹了Queue隊列中join()與task_done()的關系是什么的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇Queue隊列中join()與task_done()的關系是什么文章都會有所收獲，下面我們一起來看看吧。

join()與task_done()的關系

在網上大多關于join()與task_done()的結束原話是這樣的：

• Queue.task_done() 在完成一項工作之后，Queue.task_done()函數向任務已經完成的隊列發送一個信號 

• Queue.join() 實際上意味著等到隊列為空，再執行別的操作

理解

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

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

下面看個自己寫的例子：

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

#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以為什么要用線程，因為線程可以start后繼續執行后面的主線程，可以put數據，如果不是線程直接在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) #快速生產
for x in tasks:
 t = Mythread(que) #把同一個隊列傳入2個線程
 t.start()
 
que.join()
 
print('---success---')

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

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

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

 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

快速生產-快速消費

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

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

慢速生產-快速消費

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

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

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

代碼如下：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以為什么要用線程，因為線程可以start后繼續執行后面的主線程，可以put數據，如果不是線程直接在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() #這里要放到判斷前，否則取最后最后一個的時候已經為空，直接break，task_done執行不了，join()判斷隊列一直沒結束
 if item == None:
 break
 print(item,'!')
return
que = queue.Queue()
tasks = [Mythread(que) for x in range(1)]
 #快速生產
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---')

注意點

put隊列完成的時候千萬不能用task_done()，否則會報錯：

task_done() called too many times

因為該方法僅僅表示get成功后，執行的一個標記。

關于“Queue隊列中join()與task_done()的關系是什么”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“Queue隊列中join()與task_done()的關系是什么”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注億速云行業資訊頻道。