python queue 学习笔记

王艳玲

part 1：get help


>>> import Queue
>>> Queue.__all__
['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue']

>>> help(Queue)



NAME
Queue - A multi-producer, multi-consumer queue.

FILE
/usr/lib/python2.6/Queue.py

MODULE DOCS
http://docs.python.org/library/Queue

CLASSES
Queue
LifoQueue
PriorityQueue
exceptions.Exception(exceptions.BaseException)
Empty
Full



# 可以看到主要有5个类，Empty, Full 是两个和异常相关的类。 Queue 是基本的序列类， LifoQueue， PriorityQueue 继承了Queue.




part 2:


主要看一下 Queue


class Queue
| Create a queue object with a given maximum size.
|
| If maxsize is <= 0, the queue size is infinite.
|
| Methods defined here:
|
| __init__(self, maxsize=0)
|
| empty(self)
| Return True if the queue is empty, False otherwise (not reliable!).
|
| full(self)
| Return True if the queue is full, False otherwise (not reliable!).
|
| get(self, block=True, timeout=None)
| Remove and return an item from the queue.
|
| If optional args 'block' is true and 'timeout' is None (the default),
| block if necessary until an item is available. If 'timeout' is
| a positive number, it blocks at most 'timeout' seconds and raises
| the Empty exception if no item was available within that time.
| Otherwise ('block' is false), return an item if one is immediately
| available, else raise the Empty exception ('timeout' is ignored
| in that case).
|
| get_nowait(self)
| Remove and return an item from the queue without blocking.
|
| Only get an item if one is immediately available. Otherwise
| raise the Empty exception.

  |
| join(self)
| Blocks until all items in the Queue have been gotten and processed.
|
| The count of unfinished tasks goes up whenever an item is added to the
| queue. The count goes down whenever a consumer thread calls task_done()
| to indicate the item was retrieved and all work on it is complete.
|
| When the count of unfinished tasks drops to zero, join() unblocks.
|
| put(self, item, block=True, timeout=None)
| Put an item into the queue.
|
| If optional args 'block' is true and 'timeout' is None (the default),
| block if necessary until a free slot is available. If 'timeout' is
| a positive number, it blocks at most 'timeout' seconds and raises
| the Full exception if no free slot was available within that time.
| Otherwise ('block' is false), put an item on the queue if a free slot
| is immediately available, else raise the Full exception ('timeout'
| is ignored in that case).
|
| put_nowait(self, item)
| Put an item into the queue without blocking.
|
| Only enqueue the item if a free slot is immediately available.
| Otherwise raise the Full exception.
|
| qsize(self)
| Return the approximate size of the queue (not reliable!).
|
| 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.





Queue.Queue() 一个队列的同步实现！！
# 实例化一个Queue
实例化一个Queue 实例。调用 __init__() 方法。 默认maxsize = 0, 如果 maxsize <= 0 （maxsize < 1）, 认为是队列长度没有限制的（底层做动态扩展）
>>> myqueue = Queue()


# 查看一个Queue 是否为空
>>> myqueue.empty()
True


# 查看一个Queue 是否为满
full() 与 maxsize 对应。 queue 时候为满，主要是判断 iterm 个数时候达到 maxsize
>>> myqueue.full()
False



# 先队列中添加一个元素
缺省参数 block = True, 意为添加元素时，queue 是阻塞的，其他线程是不能操作此队列的。
如果block=False, 有一个潜在的危险： 此时向队列中添加元素，可能别的线程也会向该queue 添加线程，此时可能产生一个full exception


>>> myqueue.put('jia')
>>> myqueue.empty()
False
>>> myqueue.full()
False


# 查看队列中的元素个数
>>> myqueue.qsize()
1


# 从队头删除并返回一个iterm
# 默认参数 block = True. 如果队列为空，block


# get() 等效于 get(True,None) ， 即get() 方法一直阻塞，直到取出第一个元素为止。
# NOTE: function() 中如果某一个参数使用了缺省参数，之后的参数也要使用缺省参数；
# function() 调用时，如果对某一个缺省参数传值，必须确保该参数之前的缺省参数也得到有效赋值！除非使用key=value的形式，明确告知function() 这次的赋值是对第二个参数。否则function 会将你传入的参数按照他自己的列表顺序来解析！！


# 一个常使用的写法是： myqueue.get(True,1). 第二个参数指定最大的阻塞时间，


>>> myqueue.get()
'jia'

>>> myqueue = Queue()
>>> myqueue.put('jia')
>>> myqueue.get(timeout=1)
'jia'
>>> myqueue.get(timeout=1)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python2.6/Queue.py", line 176, in get
raise Empty
Queue.Empty
>>> myqueue.put('jia')
>>> myqueue.get(True)
'jia'



# 对get() 方法的推荐使用： 使用之前 通过 q.qsize(), q.empty() 检验队列时候为空， 确认队列非空后， 使用 q.get(True,1) 或者 q.get(timeout=1) 获取元素
# 如果queue 为空， q.ge() 会一直等下去，直到有一个iterm 可以被返回。


# put_nowait() get_nowait() 是对put() get() 的扩展。
get_nowait() 相当于 get(block=False),
put_nowait(iterm) 相当于 put(iterm, False)

>>> myqueue = Queue()
>>> myqueue.put('jia')
>>> myqueue.put_nowait('luo')
>>> myqueue.get_nowait()
'jia'
>>> myqueue.get_nowait()
'luo'


# join 似乎是和Queue 队列相关的一个操作！！
task_done() 完成一项任务后， task_done() 想任务已经完成的队列发送一个信号
join() 意味着等到队列为空，再执行别的操作。 似乎有一个队列锁的问题。




# task_done 不能调用多次！
>>> myqueue.task_done()

>>> myqueue.task_done()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python2.6/Queue.py", line 64, in task_done
raise ValueError('task_done() called too many times')
ValueError: task_done() called too many times



# think 1: Queue.qsize() == 0 和 Queue.empty() is True 有区别吗？


对三种队列的分析：

1、python queue模块的FIFO队列先进先出。
2、LIFO类似于堆。即先进后出。
3、还有一种是优先级队列级别越低越先出来。

针对这三种队列分别有三个构造函数:
1、class Queue.Queue(maxsize) FIFO
2、class Queue.LifoQueue(maxsize) LIFO
3、class Queue.PriorityQueue(maxsize) 优先级队列



# 一个 queue , threading 的使用示例。


#coding:utf-8


importQueue
importthreading
importtime
importrandom


q = Queue.Queue(0)#当有多个线程共享一个东西的时候就可以用它了
# 严格的来说，单单是队列还是很简单的。 如果和线程结合起来，就复杂了！！


NUM_WORKERS =3


classMyThread(threading.Thread):


def__init__(self,input,worktype):
self._jobq = input
self._work_type = worktype

# 一般来说： 如果子类要想使用父类的相关属性，必须在子类中显式的调用父类的init方法；
# 如果只是想 判断子类的一个实例的属性是父类，可以不用显式调用父类的init方法
threading.Thread.__init__(self)


defrun(self):
whileTrue:
# 如果队列不为空，执行操作
ifself._jobq.qsize()>0:
# 从 job list 中获取一个任务， 添加一个 work_type,
self._process_job(self._jobq.get(),self._work_type)
else:
break


def_process_job(self,job, worktype):
doJob(job,worktype)


defdoJob(job,worktype):
time.sleep(random.random() *3)
print"doing",job,"worktype",worktype


if__name__ =='__main__':
print"begin...."
# 把这些东西放到 队列中!
foriinrange(NUM_WORKERS*2):
q.put(i)#放入到任务队列中去
print"jobq size:",q.qsize()

# range() 必须接受一个 整型数据！

forxinrange(NUM_WORKERS):
# 调用线程的start() 方法
MyThread(q,x).start()