Implementing a Multithreaded Priority Queue

queueLock = threading.Lock() workQueue = Queue.Queue(10) queueLock.acquire() for word in wordList:     workQueue.put(word) queueLock.release() while not workQueue.empty():     pass . . . queueLock.acquire() if not workQueue.empty():     data = q.get()     queueLock.release()

The Queue module provides an invaluable way to manage processing large amounts of data on multiple threads. The Queue module allows you to create a new queue object that can hold a specific number of items. Items can be added and removed from the queue using the get() and put() methods of the queue object.

The queue object also includes the empty(), full(), and qsize() methods to determine whether the queue is empty, full, or the approximate size, respectively. The qsize method is not always reliable because of multiple threads removing items from the queue.

If necessary, you can implement the thread locking discussed in the previous phrase to control access to the queue. This will make queue management much safer and provide you with more control of the data processing.

import Queue import threading import time import thread doExit = 0 class newThread (threading.Thread):     def __init__(self, threadID, name, q):         self.threadID = threadID         self.name = name         self.q = q         threading.Thread.__init__(self)     def run(self):         print "Starting " + self.name         process_data(self.name, self.q)         print "Exiting " + self.name def process_data(tName, q):     while not doExit:         queueLock.acquire()         if not workQueue.empty():             data = q.get()             queueLock.release()             print "%s processing %s" % (tName, data)         else:             queueLock.release()         time.sleep(1) threadList = ["Thread1", "Thread2", "Thread3"] wordList = ["One", "Two", "Three", "Four", "Five"] queueLock = threading.Lock() workQueue = Queue.Queue(10) threads = [] tID = 1 #Create new threads for tName in threadList:     thread = newThread(tID, tName, workQueue)     thread.start()     threads.append(thread)     tID += 1 #Fill the queue queueLock.acquire() for word in wordList:     workQueue.put(word) queueLock.release() #Wait for queue to empty while not workQueue.empty():     pass #Notify threads it's time to exit doExit = 1 #Wait for all threads to complete for t in threads:     t.join() print "Exiting Main Thread"

queue_thread.py

Starting Thread1 Starting Thread2 Starting Thread3 Thread1 processing One Thread2 processing Two Thread3 processing Three Thread1 processing Four Thread2 processing Five Exiting Thread1 Exiting Thread2 Exiting Thread3 Exiting Main Thread

Output from queue_thread.py code