The Queue Module

The Queue module provides a thread-safe queue implementation, shown in Example 3-2. It provides a convenient way of moving Python objects between different threads.

Example 3-2. Using the Queue Module

File: queue-example-1.py

import threading
import Queue
import time, random

WORKERS = 2

class Worker(threading.Thread):

 def _ _init_ _(self, queue):
 self._ _queue = queue
 threading.Thread._ _init_ _(self)

 def run(self):
 while 1:
 item = self._ _queue.get()
 if item is None:
 break # reached end of queue

 # pretend we're doing something that takes 10100 ms
 time.sleep(random.randint(10, 100) / 1000.0)

 print "task", item, "finished"

#
# try it

queue = Queue.Queue(0)

for i in range(WORKERS):
 Worker(queue).start() # start a worker

for i in range(10):
 queue.put(i)

for i in range(WORKERS):
 queue.put(None) # add end-of-queue markers

task 1 finished
task 0 finished
task 3 finished
task 2 finished
task 4 finished
task 5 finished
task 7 finished
task 6 finished
task 9 finished
task 8 finished

Example 3-3 shows how you can limit the size of the queue. If the producer threads fill the queue, they will block until items are popped off the queue.

Example 3-3. Using the Queue Module with a Maximum Size

File: queue-example-2.py

import threading
import Queue

import time, random

WORKERS = 2

class Worker(threading.Thread):

 def _ _init_ _(self, queue):
 self._ _queue = queue
 threading.Thread._ _init_ _(self)

 def run(self):
 while 1:
 item = self._ _queue.get()
 if item is None:
 break # reached end of queue

 # pretend we're doing something that takes 10100 ms
 time.sleep(random.randint(10, 100) / 1000.0)

 print "task", item, "finished"

#
# run with limited queue

queue = Queue.Queue(3)

for i in range(WORKERS):
 Worker(queue).start() # start a worker

for item in range(10):
 print "push", item
 queue.put(item)

for i in range(WORKERS):
 queue.put(None) # add end-of-queue markers

push 0
push 1
push 2
push 3
push 4
push 5
task 0 finished
push 6
task 1 finished
push 7
task 2 finished
push 8
task 3 finished
push 9
task 4 finished
task 6 finished
task 5 finished
task 7 finished
task 9 finished
task 8 finished

You can modify the behavior through subclassing. The class in Example 3-4 provides a simple priority queue. It expects all items added to the queue to be tuples, where the first member contains the priority (lower value means higher priority).

Example 3-4. Using the Queue Module to Implement a Priority Queue

File: queue-example-3.py

import Queue
import bisect

Empty = Queue.Empty

class PriorityQueue(Queue.Queue):
 "Thread-safe priority queue"

 def _put(self, item):
 # insert in order
 bisect.insort(self.queue, item)

#
# try it

queue = PriorityQueue(0)

# add items out of order
queue.put((20, "second"))
queue.put((10, "first"))
queue.put((30, "third"))

# print queue contents
try:
 while 1:
 print queue.get_nowait()
except Empty:
 pass

third
second
first

Example 3-5 shows a simple stack implementation (last-in, first-out, instead of first-in, first-out).

Example 3-5. Using the Queue Module to Implement a Stack

File: queue-example-4.py

import Queue

Empty = Queue.Empty

class Stack(Queue.Queue):
 "Thread-safe stack"
 
 def _put(self, item):
 # insert at the beginning of queue, not at the end
 self.queue.insert(0, item)

 # method aliases
 push = Queue.Queue.put
 pop = Queue.Queue.get
 pop_nowait = Queue.Queue.get_nowait

#
# try it

stack = Stack(0)

# push items on stack
stack.push("first")
stack.push("second")
stack.push("third")

# print stack contents
try:
 while 1:
 print stack.pop_nowait()
except Empty:
 pass

third
second
first