queue库-Queue源码学习

 1 class Queue:
 2 
 3     def __init__(self, maxsize=0):
 4         self.maxsize = maxsize
 5         self._init(maxsize)
 6 
 7         # mutex must be held whenever the queue is mutating.  All methods
 8         # that acquire mutex must release it before returning.  mutex
 9         # is shared between the three conditions, so acquiring and
10         # releasing the conditions also acquires and releases mutex.
11         self.mutex = threading.Lock()
12 
13         # Notify not_empty whenever an item is added to the queue; a
14         # thread waiting to get is notified then.
15         self.not_empty = threading.Condition(self.mutex)
16 
17         # Notify not_full whenever an item is removed from the queue;
18         # a thread waiting to put is notified then.
19         self.not_full = threading.Condition(self.mutex)
20 
21         # Notify all_tasks_done whenever the number of unfinished tasks
22         # drops to zero; thread waiting to join() is notified to resume
23         self.all_tasks_done = threading.Condition(self.mutex)
24         self.unfinished_tasks = 0

 1 def put(self, item, block=True, timeout=None):
 2     with self.not_full:
 3         if self.maxsize > 0:
 4             ...
 5             elif timeout is None:
 6                 while self._qsize() >= self.maxsize:
 7                     self.not_full.wait()
 8             ...
 9         self._put(item)
10         self.unfinished_tasks += 1
11         self.not_empty.notify()
12 
13 
14 def get(self, block=True, timeout=None):
15     with self.not_empty:
16         ...
17         elif timeout is None:
18             while not self._qsize():
19                 self.not_empty.wait()
20         ...
21         item = self._get()
22         self.not_full.notify()
23         return item

 1 def join(self):
 2     with self.all_tasks_done:
 3         while self.unfinished_tasks:
 4             self.all_tasks_done.wait()
 5 
 6 def task_done(self):
 7     with self.all_tasks_done:
 8         unfinished = self.unfinished_tasks - 1
 9         if unfinished <= 0:
10             if unfinished < 0:
11                 raise ValueError(‘task_done() called too many times‘)
12             self.all_tasks_done.notify_all()
13         self.unfinished_tasks = unfinished