理解Queue队列中join()与task_done()的关系

在网上大多关于join()与task_done()的结束原话是这样的：

但是可能很多人还是不太理解，这里以我自己的理解来阐述这两者的关联。

理解

如果线程里每从队列里取一次，但没有执行task_done()，则join无法判断队列到底有没有结束，在最后执行个join()是等不到结果的，会一直挂起。

可以理解为，每task_done一次 就从队列里删掉一个元素，这样在最后join的时候根据队列长度是否为零来判断队列是否结束，从而执行主线程。

下面看个自己写的例子：

下面这个例子，会在join()的地方无限挂起，因为join在等队列清空，但是由于没有task_done，它认为队列还没有清空，还在一直等。

 

1.  
   #!/usr/bin/env python
2.  
   # -*- coding:utf-8 -*-
3.  
   '''threading test'''
4.  
   import threading
5.  
   import queue
6.  
   from time import sleep
7.  
   #之所以为什么要用线程，因为线程可以start后继续执行后面的主线程，可以put数据，如果不是线程直接在get阻塞。
8.  
   class Mythread(threading.Thread):
9.  
   def __init__(self,que):
10.  
    threading.Thread.__init__(self)
11.  
    self.queue = que
12.  
    def run(self):
13.  
    while True:
14.  
    sleep(1)
15.  
    if self.queue.empty(): #判断放到get前面，这样可以，否则队列最后一个取完后就空了，直接break，走不到print
16.  
    break
17.  
    item = self.queue.get()
18.  
    print(item,'!')
19.  
    #self.queue.task_done()
20.  
    return
21.  
    que = queue.Queue()
22.  
    tasks = [Mythread(que) for x in range(1)]
23.  
    for x in range(10):
24.  
     
25.  
    que.put(x) #快速生产
26.  
    for x in tasks:
27.  
    t = Mythread(que) #把同一个队列传入2个线程
28.  
    t.start()
29.  
     
30.  
    que.join()
31.  
     
32.  
    print('---success---')
33.  
     

如果把self.queue.task_done()  注释去掉，就会顺利执行完主程序。

这就是“Queue.task_done()函数向任务已经完成的队列发送一个信号”这句话的意义，能够让join()函数能判断出队列还剩多少，是否清空了。

而事实上我们看下queue的源码可以看出确实是执行一次未完成队列减一：

1.  
   def task_done(self):
2.  
   '''Indicate that a formerly enqueued task is complete.
3.  
    
4.  
   Used by Queue consumer threads. For each get() used to fetch a task,
5.  
   a subsequent call to task_done() tells the queue that the processing
6.  
   on the task is complete.
7.  
    
8.  
   If a join() is currently blocking, it will resume when all items
9.  
   have been processed (meaning that a task_done() call was received
10.  
    for every item that had been put() into the queue).
11.  
     
12.  
    Raises a ValueError if called more times than there were items
13.  
    placed in the queue.
14.  
    '''
15.  
    with self.all_tasks_done:
16.  
    unfinished = self.unfinished_tasks - 1
17.  
    if unfinished <= 0:
18.  
    if unfinished < 0:
19.  
    raise ValueError('task_done() called too many times')
20.  
    self.all_tasks_done.notify_all()
21.  
    self.unfinished_tasks = unfinished
22.  
     
23.  
     

快速生产-快速消费

上面的演示代码是快速生产-慢速消费的场景，我们可以直接用task_done()与join()配合，来让empty()判断出队列是否已经结束。 当然，queue我们可以正确判断是否已经清空，但是线程里的get队列是不知道，如果没有东西告诉它，队列空了，因此get还会继续阻塞，那么我们就需要在get程序中加一个判断，如果empty()成立，break退出循环，否则get()还是会一直阻塞。

慢速生产-快速消费

但是如果生产者速度与消费者速度相当，或者生产速度小于消费速度，则靠task_done()来实现队列减一则不靠谱，队列会时常处于供不应求的状态，常为empty，所以用empty来判断则不靠谱。 那么这种情况会导致 join可以判断出队列结束了，但是线程里不能依靠empty()来判断线程是否可以结束。 我们可以在消费队列的每个线程最后塞入一个特定的“标记”，在消费的时候判断，如果get到了这么一个“标记”，则可以判定队列结束了，因为生产队列都结束了，也不会再新增了。 代码如下：

1.  
   #!/usr/bin/env python
2.  
   # -*- coding:utf-8 -*-
3.  
   '''threading test'''
4.  
   import threading
5.  
   import queue
6.  
   from time import sleep
7.  
   #之所以为什么要用线程，因为线程可以start后继续执行后面的主线程，可以put数据，如果不是线程直接在get阻塞。
8.  
   class Mythread(threading.Thread):
9.  
   def __init__(self,que):
10.  
    threading.Thread.__init__(self)
11.  
    self.queue = que
12.  
    def run(self):
13.  
    while True:
14.  
    item = self.queue.get()
15.  
    self.queue.task_done() #这里要放到判断前，否则取最后最后一个的时候已经为空，直接break，task_done执行不了，join()判断队列一直没结束
16.  
    if item == None:
17.  
    break
18.  
    print(item,'!')
19.  
    return
20.  
    que = queue.Queue()
21.  
    tasks = [Mythread(que) for x in range(1)]
22.  
    #快速生产
23.  
    for x in tasks:
24.  
    t = Mythread(que) #把同一个队列传入2个线程
25.  
    t.start()
26.  
    for x in range(10):
27.  
    sleep(1)
28.  
    que.put(x)
29.  
    for x in tasks:
30.  
    que.put(None)
31.  
    que.join()
32.  
    print('---success---')

注意点：

put队列完成的时候千万不能用task_done()，否则会报错：

task_done() called too many times

因为该方法仅仅表示get成功后，执行的一个标记。