Queue模塊及源碼分析

from time import time as _time
try:
 import threading as _threading
except ImportError:
 import dummy_threading as _threading
from collections import deque
import heapq
 
__all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue']
 
class Empty(Exception):
 "Exception raised by Queue.get(block=0)/get_nowait()."
 pass
 
class Full(Exception):
 "Exception raised by Queue.put(block=0)/put_nowait()."
 pass
 
class Queue:
 def __init__(self, maxsize=0):
 self.maxsize = maxsize
 self._init(maxsize)
 self.mutex = _threading.Lock()
 self.not_empty = _threading.Condition(self.mutex)
 self.not_full = _threading.Condition(self.mutex)
 self.all_tasks_done = _threading.Condition(self.mutex)
 self.unfinished_tasks = 
 
 def get_nowait(self):
 return self.get(False)
 def _init(self, maxsize):
 self.queue = deque()
 def _qsize(self, len=len):
 return len(self.queue)
 def _put(self, item):
 self.queue.append(item)
 def _get(self):
 return self.queue.popleft()

class LifoQueue(Queue):
 '''Variant of Queue that retrieves most recently added entries first.'''
 
 def _init(self, maxsize):
 self.queue = []
 
 def _qsize(self, len=len):
 return len(self.queue)
 
 def _put(self, item):
 self.queue.append(item)
 
 def _get(self):
 return self.queue.pop()

class PriorityQueue(Queue):
 '''Variant of Queue that retrieves open entries in priority order (lowest first).
 
 Entries are typically tuples of the form: (priority number, data).
 '''
 
 def _init(self, maxsize):
 self.queue = []
 
 def _qsize(self, len=len):
 return len(self.queue)
 
 def _put(self, item, heappush=heapq.heappush):
 heappush(self.queue, item)
 
 def _get(self, heappop=heapq.heappop):
 return heappop(self.queue)

def put(self, item, block=True, timeout=None):
 self.not_full.acquire() #not_full獲得鎖
 try:
 if self.maxsize > 0: #如果隊列長度有限制
 if not block: #如果沒阻塞
 if self._qsize() == self.maxsize: #如果隊列滿了拋異常
 raise Full
 elif timeout is None: #有阻塞且超時為空，等待
 while self._qsize() == self.maxsize:
 self.not_full.wait()
 elif timeout < 0:
 raise ValueError("'timeout' must be a non-negative number")
 else: #如果有阻塞，且超時非負時，結束時間=當前時間+超時時間
 endtime = _time() + timeout
 while self._qsize() == self.maxsize:
 remaining = endtime - _time()
 if remaining <= 0.0: #到時后，拋異常
 raise Full
 #如果沒到時，隊列是滿的就會一直被掛起，直到有“位置”騰出
 self.not_full.wait(remaining)
 self._put(item) #調用_put方法，添加元素
 self.unfinished_tasks += 1 #未完成任務+1
 self.not_empty.notify() #通知非空，喚醒非空掛起的任務
 finally:
 self.not_full.release() #not_full釋放鎖

def get(self, block=True, timeout=None):
 
 self.not_empty.acquire() #not_empty獲得鎖
 try:
 if not block: #不阻塞時
 if not self._qsize(): #隊列為空時拋異常
 raise Empty
 elif timeout is None: #不限時時，隊列為空則會等待
 while not self._qsize():
 self.not_empty.wait()
 elif timeout < 0:
 raise ValueError("'timeout' must be a non-negative number")
 else:
 endtime = _time() + timeout
 while not self._qsize():
 remaining = endtime - _time()
 if remaining <= 0.0:
 raise Empty
 self.not_empty.wait(remaining)
 item = self._get() #調用_get方法，移除并獲得項目
 self.not_full.notify() #通知非滿
 return item #返回項目
 finally:
 self.not_empty.release() #釋放鎖

def task_done(self):
 
 self.all_tasks_done.acquire() #獲得鎖
 try:
 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 #否則未完成任務數量-1
 finally:
 self.all_tasks_done.release() #最后釋放鎖

def join(self):
 
 self.all_tasks_done.acquire()
 try:
 while self.unfinished_tasks: #如果有未完成的任務，將調用wait()方法等待
 self.all_tasks_done.wait()
 finally:
 self.all_tasks_done.release()

import random,threading,time
from Queue import Queue
is_product = True
class Producer(threading.Thread):
 """生產數據"""
 def __init__(self, t_name, queue):
 threading.Thread.__init__(self,name=t_name)
 self.data=queue
 def run(self):
 while 1:
 
 if self.data.full():
 global is_product
 is_product = False
 else:
 if self.data.qsize() <= 7:#隊列長度小于等于7時添加元素
 is_product = True
 for i in range(2): #每次向隊列里添加兩個元素
 
 randomnum=random.randint(1,99)
 print "%s: %s is producing %d to the queue!" % (time.ctime(), self.getName(), randomnum)
 self.data.put(randomnum，False) #將數據依次存入隊列
 time.sleep(1)
 print "deque length is %s"%self.data.qsize()
 else:
 if is_product:
 for i in range(2): #
 
 randomnum = random.randint(1, 99)
 print "%s: %s is producing %d to the queue!" % (time.ctime(), self.getName(), randomnum)
 self.data.put(randomnum,False) # 將數據依次存入隊列
 time.sleep(1)
 print "deque length is %s" % self.data.qsize()
 else:
 pass
 
 print "%s: %s finished!" %(time.ctime(), self.getName())
 
#Consumer thread
class Consumer_even(threading.Thread):
 def __init__(self,t_name,queue):
 threading.Thread.__init__(self,name=t_name)
 self.data=queue
 def run(self):
 while 1:
 if self.data.qsize() > 7:#隊列長度大于7時開始取元素
 val_even = self.data.get(False)
 if val_even%2==0:
 print "%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(),self.getName(),val_even)
 time.sleep(2)
 else:
 self.data.put(val_even)
 time.sleep(2)
 print "deque length is %s" % self.data.qsize()
 else:
 pass
 
 
class Consumer_odd(threading.Thread):
 def __init__(self,t_name,queue):
 threading.Thread.__init__(self, name=t_name)
 self.data=queue
 def run(self):
 while 1:
 if self.data.qsize() > 7:
 val_odd = self.data.get(False)
 if val_odd%2!=0:
 print "%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(), self.getName(), val_odd)
 time.sleep(2)
 else:
 self.data.put(val_odd)
 time.sleep(2)
 print "deque length is %s" % self.data.qsize()
 else:
 pass
 
#Main thread
def main():
 queue = Queue(20)
 producer = Producer('Pro.', queue)
 consumer_even = Consumer_even('Con_even.', queue)
 consumer_odd = Consumer_odd('Con_odd.',queue)
 producer.start()
 consumer_even.start()
 consumer_odd.start()
 producer.join()
 consumer_even.join()
 consumer_odd.join()
 
if __name__ == '__main__':
 main()