一个以前写的生产者－消费者的线程代码的启示

今天研究了一个以前的写的生产者－消费者的线程代码，这个代码会一直hang下去. 代码如下

点击(此处)折叠或打开

1. # two entities here that try to share a common resource, a queue
   
2. import threading
   
3. from threading import Thread,Event
   
4. from queue import Queue
   
5. import time,random
   
6. import logging
   
7. import sys
   
8. 
   
9. logging.basicConfig(level=logging.DEBUG,format='%(asctime)s (%(threadName)-10s) %(message)s',)
   
10. _sentinel=object()
    
11. #_sentinel = None
    
12. 
    
13. class Producer(threading.Thread):
    
14.     def __init__(self,queue):
    
15.         Thread.__init__(self)
    
16.         self.queue=queue
    
17.         self.daemon=True
    
18. 
    
19.     def run(self):
    
20.         for i in range(5):
    
21.             item=random.randint(0,256)
    
22.             self.queue.put(item)
    
23.             logging.debug('Producer notify: item %d appended to queue by %s\n' %(item,self.name))
    
24.         self.queue.put(_sentinel)
    
25.         
    
26. 
    
27. class Consumer(threading.Thread):
    
28.     def __init__(self,queue):
    
29.         Thread.__init__(self)
    
30.         self.queue=queue
    
31.         self.daemon=True
    
32.     
    
33.     def run(self):
    
34.         while True:
    
35.             item=self.queue.get()
    
36.             if item is _sentinel:
    
37.                 self.queue.put(_sentinel)
    
38.                 break
    
39.             else:
    
40.                 logging.debug('Consumer notify: %d poped from queue by %s' %(item,self.name))
    
41.                 self.queue.task_done()
    
42. 
    
43. def main():
    
44.     shared_queue=Queue()
    
45.     consumers = []
    
46.     #start the producer
    
47.     p1=Producer(shared_queue)
    
48.     p1.start()
    
49.    
    
50.     #start 3 consumer
    
51.     for i in range(0, 3):
    
52.         t=Consumer(shared_queue)
    
53.         t.start()
    
54.         consumers.append(t)
    
55. 
    
56.      
    
57.     #join the queue
    
58.     shared_queue.join()
    
59.     
    
60.     #stop producer
    
61.     p1.join()
    
62.     
    
63.     #stop consumers
    
64.     for t in consumers:
    
65.         t.join()
    
66. 
    
67. if __name__=='__main__':
    
68.     main()

点击(此处)折叠或打开

1. root@kali:/usr/local/src/py/network# ./thread_share_queue_hang.py
   
2. 2018-09-05 20:20:01,481 (Thread-1 ) Producer notify: item 93 appended to queue by Thread-1
   
3. 
   
4. 2018-09-05 20:20:01,482 (Thread-1 ) Producer notify: item 144 appended to queue by Thread-1
   
5. 
   
6. 2018-09-05 20:20:01,482 (Thread-2 ) Consumer notify: 93 poped from queue by Thread-2
   
7. 2018-09-05 20:20:01,482 (Thread-1 ) Producer notify: item 45 appended to queue by Thread-1
   
8. 
   
9. 2018-09-05 20:20:01,482 (Thread-2 ) Consumer notify: 144 poped from queue by Thread-2
   
10. 2018-09-05 20:20:01,482 (Thread-3 ) Consumer notify: 45 poped from queue by Thread-3
    
11. 2018-09-05 20:20:01,482 (Thread-1 ) Producer notify: item 106 appended to queue by Thread-1
    
12. 
    
13. 2018-09-05 20:20:01,483 (Thread-4 ) Consumer notify: 106 poped from queue by Thread-4
    
14. 2018-09-05 20:20:01,483 (Thread-1 ) Producer notify: item 219 appended to queue by Thread-1
    
15. 
    
16. 2018-09-05 20:20:01,483 (Thread-2 ) Consumer notify: 219 poped from queue by Thread-2
    
17. ^CTraceback (most recent call last):
    
18.   File "./thread_share_queue_hang.py", line 73, in <module>
    
19.     main()
    
20.   File "./thread_share_queue_hang.py", line 63, in main
    
21.     shared_queue.join()

这段代码会hang在63行，shared_queue.join().  为什么呢？看官方文档的解释，join方法会阻塞直到queue里面所有的item都被处理完。producer 和３个consumer 都往queue里面放了一个None, 共有４个None, 而其中三个None被用来做sentinel 来使得３个consumer线程能够结束。最后留下一个None在queue里面，而没有人来处理最后的这个None,所以会导致queue.join()会一直hang.

点击(此处)折叠或打开

1. Queue.join()
   
2. 
   
3.     Blocks until all items in the queue have been gotten and processed.

那么问题找到了，我就重新写了这个main()部分，但是我用了不同的思路:
第一种：
main thread来检查是否只剩下最后一个线程了，如果只剩下最后一个线程(即它自己)
就从queue里面取出最后一个元素处理它.用task_done()函数。

点击(此处)折叠或打开

1. def main():
   
2.     shared_queue=Queue()
   
3.     consumers = []
   
4.     #start the producer
   
5.     p1=Producer(shared_queue)
   
6.     p1.start()
   
7.    
   
8.     #start 3 consumer
   
9.     for i in range(0, 3):
   
10.         t=Consumer(shared_queue)
    
11.         t.start()
    
12.         consumers.append(t)
    
13. 
    
14.      
    
15.     while True:
    
16.         #if I - the main thread also am the last thread
    
17.         #print(threading.active_count())
    
18.         if threading.active_count() == 1:
    
19.             #get the last elem from the queue
    
20.             if shared_queue.get() == _sentinel:
    
21.                 shared_queue.task_done()
    
22.                 break
    
23.         else:
    
24.             time.sleep(0.5)

第二种:
使用qsize() 方法来检查如果queue里面只剩下一个elem,那么我处理它就可以了。
因为这时候其他线程都已经结束了。但是这个qsize() 根据文档来说并不可靠。not reliable

点击(此处)折叠或打开

1. #create shared queue
   
2.     shared_queue=Queue()
   
3.     consumers = []
   
4. 
   
5.     #start the producer
   
6.     p1=Thread(target=producer, args=(shared_queue, ))
   
7.     p1.start()
   
8.    
   
9.     #start 3 consumer
   
10.     for i in range(0, 3):
    
11.         t=Thread(target=consumer, args=(shared_queue,))
    
12.         t.start()
    
13.         consumers.append(t)
    
14. 
    
15.      
    
16.     while True:
    
17.         #print(threading.active_count())
    
18.         #qsize() method of queue.Queue instance
    
19.         #Return the approximate size of the queue (not
    
20.         #use the qsize() method to justify whether there is only 1 None left in the shared queue
    
21.         if shared_queue.qsize() == 1:
    
22.             #get the last elem from the queue
    
23.             if shared_queue.get_nowait() is None:
    
24.                 shared_queue.task_done()
    
25.                 break
    
26.         else:
    
27.             time.sleep(0.5)

第三种:
干脆我不对queue做join操作，只要我的线程做join就可以了。

点击(此处)折叠或打开

1. shared_queue=Queue()
   
2.     consumers = []
   
3.     #start the producer
   
4.     p1=Producer(shared_queue)
   
5.     p1.start()
   
6.    
   
7.     #start 3 consumer
   
8.     for i in range(0, 3):
   
9.         t=Consumer(shared_queue)
   
10.         t.start()
    
11.         consumers.append(t)
    
12. 
    
13.     #block until all task in shared_queue are done
    
14.     #we can not do this, because we put 4 None in the queue, but only use 3 of them
    
15.     #as sentinel for consumer, thus use shared_queue.join() will block forever
    
16.     #shared_queue.join()
    
17.     
    
18.     #stop producer
    
19.     p1.join()
    
20.     
    
21.     #stop consumers
    
22.     for t in consumers:
    
23.         t.join()

这三种写法都达到了同样的效果，但是重新看了queue的文档，让我感觉学到了新知识。