Multithreading - Producer and consumer with Queue

Python Multithread

In this chapter, we'll implement another version of Producer and Consumer code with Queue (see Condition objects with producer and consumer).

In the following example, the Consumer and Producer threads runs indefinitely while checking the status of the queue. The Producer thread is responsible for putting items into the queue if it is not full while the Consumer thread consumes items if there are any.

import threading
import time
import logging
import random
import Queue

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)

BUF_SIZE = 10
q = Queue.Queue(BUF_SIZE)

class ProducerThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, verbose=None):
        super(ProducerThread,self).__init__()
        self.target = target
        self.name = name

    def run(self):
        while True:
            if not q.full():
                item = random.randint(1,10)
                q.put(item)
                logging.debug('Putting ' + str(item)  
                              + ' : ' + str(q.qsize()) + ' items in queue')
                time.sleep(random.random())
        return

class ConsumerThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, verbose=None):
        super(ConsumerThread,self).__init__()
        self.target = target
        self.name = name
        return

    def run(self):
        while True:
            if not q.empty():
                item = q.get()
                logging.debug('Getting ' + str(item) 
                              + ' : ' + str(q.qsize()) + ' items in queue')
                time.sleep(random.random())
        return

if __name__ == '__main__':
    
    p = ProducerThread(name='producer')
    c = ConsumerThread(name='consumer')

    p.start()
    time.sleep(2)
    c.start()
    time.sleep(2)

Output:

(producer ) Putting 2 : 1 items in queue
(producer ) Putting 10 : 2 items in queue
(producer ) Putting 6 : 3 items in queue
(producer ) Putting 7 : 4 items in queue
(producer ) Putting 1 : 5 items in queue
(consumer ) Getting 2 : 4 items in queue
(consumer ) Getting 10 : 3 items in queue
(producer ) Putting 1 : 4 items in queue
(producer ) Putting 8 : 5 items in queue
(consumer ) Getting 6 : 4 items in queue
(producer ) Putting 10 : 5 items in queue
...

1. Since the Queue has a Condition and that condition has its Lock we don't need to bother about Condition and Lock.

Producer uses Queue.put(item[, block[, timeout]]) to insert data in the queue. It has the logic to acquire the lock before inserting data in 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).

2. Also put() checks whether the queue is full, then it calls wait() internally and so producer starts waiting.
3. Consumer uses Queue.get([block[, timeout]]), and it acquires the lock before removing data from queue. If the queue is empty, it puts consumer in waiting state.
4. Queue.get() and Queue.get() has notify() method.

Python Multithread