Learning Queue module in python (how to run it)

Question

Recently, a queue design has been introduced regarding the ability to defer processing, as well as implement FIFO, etc.

I looked through the documentation, trying to get an approximate queue in order to understand how to implement it in my own project / program. But I have problems running this code:

import queue def worker(): while True: item = q.get() do_work(item) q.task_done() def main(): q = queue.Queue(maxsize=0) for i in range(num_worker_threads): t = Thread(target=worker) t.daemon = True t.start() for item in source(): q.put(item) q.join() # block until all tasks are done main() 

Question: I would like someone to explain what the for loops do, I get an error when I run the code, so I need to miss something.

Problem: NameError: global name 'num_worker_threads' not defined

Thank you from the newbie - Python -

Answer 1

The for loop starts a series of worker threads to perform the function defined by the worker. Here is the working code that should work on your system in python 2.7.

 import Queue import threading # input queue to be processed by many threads q_in = Queue.Queue(maxsize=0) # output queue to be processed by one thread q_out = Queue.Queue(maxsize=0) # number of worker threads to complete the processing num_worker_threads = 10 # process that each worker thread will execute until the Queue is empty def worker(): while True: # get item from queue, do work on it, let queue know processing is done for one item item = q_in.get() q_out.put(do_work(item)) q_in.task_done() # squares a number and returns the number and its square def do_work(item): return (item,item*item) # another queued thread we will use to print output def printer(): while True: # get an item processed by worker threads and print the result. Let queue know item has been processed item = q_out.get() print "%d squared is : %d" % item q_out.task_done() # launch all of our queued processes def main(): # Launches a number of worker threads to perform operations using the queue of inputs for i in range(num_worker_threads): t = threading.Thread(target=worker) t.daemon = True t.start() # launches a single "printer" thread to output the result (makes things neater) t = threading.Thread(target=printer) t.daemon = True t.start() # put items on the input queue (numbers to be squared) for item in range(10): q_in.put(item) # wait for two queues to be emptied (and workers to close) q_in.join() # block until all tasks are done q_out.join() print "Processing Complete" main() 

Python 3 version for @handle

 import queue import threading # input queue to be processed by many threads q_in = queue.Queue(maxsize=0) # output queue to be processed by one thread q_out = queue.Queue(maxsize=0) # number of worker threads to complete the processing num_worker_threads = 10 # process that each worker thread will execute until the Queue is empty def worker(): while True: # get item from queue, do work on it, let queue know processing is done for one item item = q_in.get() q_out.put(do_work(item)) q_in.task_done() # squares a number and returns the number and its square def do_work(item): return (item,item*item) # another queued thread we will use to print output def printer(): while True: # get an item processed by worker threads and print the result. Let queue know item has been processed item = q_out.get() print("{0[0]} squared is : {0[1]}".format(item) ) q_out.task_done() # launch all of our queued processes def main(): # Launches a number of worker threads to perform operations using the queue of inputs for i in range(num_worker_threads): t = threading.Thread(target=worker) t.daemon = True t.start() # launches a single "printer" thread to output the result (makes things neater) t = threading.Thread(target=printer) t.daemon = True t.start() # put items on the input queue (numbers to be squared) for item in range(10): q_in.put(item) # wait for two queues to be emptied (and workers to close) q_in.join() # block until all tasks are done q_out.join() print( "Processing Complete" ) main() 

Answer 2

You can think of the number of workflows as the number of bank counters in a bank. Thus, people (your items) stand in line (your turn), which will be processed by the bank teller (your work flow). Queues are actually a simple and straightforward mechanism for managing thread complexity.

I adjusted your code a bit to show how it works.

 import queue import time from threading import Thread def do_work(item): print("processing", item) def source(): item = 1 while True: print("starting", item) yield item time.sleep(0.2) item += 1 def worker(): while True: item = q.get() do_work(item) q.task_done() q = queue.Queue(maxsize=0) def main(): for i in range(2): t = Thread(target=worker) t.daemon = True t.start() for item in source(): q.put(item) q.join() # block until all tasks are done main()