Implement Multithreaded Queue with Python

Introduction..

In this example, we will create a task queue that holds all the tasks to be executed and a thread pool that interacts with the queue to process its elements individually.

We will begin with the question, what is a Queue?. A queue is a data structure that is a collection of different elements maintained in a very specific order. Let me explain by taking a real life example.

Assume you stand in line to pay your grocery billat a grocery shop counter, (don't ask me which grocery shop)

In a line of people waiting to pay their bills, you will notice the following:

1. People enter at one end of the line and exit from the other end.

2. If person A enters the line before person B, person A will leave the line before person B (unless person B is a celebrity or has more priority).

3. Once everyone has paid their bills, there will be no one left in the line.

Well, back to programming where a queue works in a similar fashion.

1. enqueue - Elements added to the end of the queue.

2. dequeue - Elements removed from the beginning of the queue.

There is more, First In First Out (FIFO) - elements that are added first will be removed first. Last In First Out (LIFO) - last element that is added will be removed first.

How does Python implement Queue data structure?

The queue module in Python provides a simple implementation of the queue data structure. Each queue can have the following methods.

• get(): returns the next element.

• put(): adds a new element.

• qsize(): number of current elements in queue.

• empty(): returns a Boolean, indicating whether the queue is empty.

• full(): returns a Boolean, indicating whether the queueis full.

1. We will create a function which takes an argument x then iterates through the numbers between 1 and itself(x), to perform multiplication. For e.g. when you pass 5 to this function it iterates through 1 to 5 and keep multiplying i.e. 1 times 5, 2 times 5, 3 times 5, 4 times 5, 5 times 5 finally returning the values as a list.

Example

def print_multiply(x):
output_value = []
for i in range(1, x + 1):
output_value.append(i * x)
print(f"Output \n *** The multiplication result for the {x} is - {output_value}")
print_multiply(5)

Output

*** The multiplication result for the 5 is - [5, 10, 15, 20, 25]

2. We will write another function called process_queue() which will attempt to obtain the next element of the queue object. The logic for this quite simple, keep passing the elements until the queue is empty. I will use sleep to delay proceeding a bit.

Example

def process_queue():
while True:
try:
value = my_queue.get(block=False)
except queue.Empty:
return
else:
print_multiply(value)
time.sleep(2)

3. Create a class, when a new instance is initialized and started, the process_queue() function will be called.

Example

class MultiThread(threading.Thread):
def __init__(self, name):
threading.Thread.__init__(self)
self.name = name

def run(self):
print(f" ** Starting the thread - {self.name}")
process_queue()
print(f" ** Completed the thread - {self.name}")

4. Finally, we will pass the input list of numbers and fill the queue.

# setting up variables
input_values = [5, 10, 15, 20]

# fill the queue
my_queue = queue.Queue()
for x in input_values:
my_queue.put(x)

5. Finally, putting all together.

import queue
import threading
import time

# Class
class MultiThread(threading.Thread):
def __init__(self, name):
threading.Thread.__init__(self)
self.name = name

def run(self):
print(f"Output \n ** Starting the thread - {self.name}")
process_queue()
print(f" ** Completed the thread - {self.name}")

# Process thr queue
def process_queue():
while True:
try:
value = my_queue.get(block=False)
except queue.Empty:
return
else:
print_multiply(value)
time.sleep(2)

# function to multiply
def print_multiply(x):
output_value = []
for i in range(1, x + 1):
output_value.append(i * x)
print(f" \n *** The multiplication result for the {x} is - {output_value}")

# Input variables
input_values = [2, 4, 6, 5,10,3]

# fill the queue
my_queue = queue.Queue()
for x in input_values:
my_queue.put(x)
# initializing and starting 3 threads
thread1 = MultiThread('First')
thread2 = MultiThread('Second')
thread3 = MultiThread('Third')
thread4 = MultiThread('Fourth')

# Start the threads
thread1.start()
thread2.start()
thread3.start()
thread4.start()

# Join the threads
thread1.join()
thread2.join()
thread3.join()
thread4.join()

Output

** Starting the thread - First
*** The multiplication result for the 2 is - [2, 4]

Output

** Starting the thread - Second
*** The multiplication result for the 4 is - [4, 8, 12, 16]

Output

** Starting the thread - Third
*** The multiplication result for the 6 is - [6, 12, 18, 24, 30, 36]

Output

** Starting the thread - Fourth
*** The multiplication result for the 5 is - [5, 10, 15, 20, 25]
*** The multiplication result for the 10 is - [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
*** The multiplication result for the 3 is - [3, 6, 9] ** Completed the thread - Third
** Completed the thread - Fourth
** Completed the thread - Second ** Completed the thread - First

6. We have successfully implemented queue concept. See, we have 4 threads but there are 6 values to process, so whoever comes first to the Queue will be executed and others will be in line to wait for others to complete.

This is similar to a real life, assume there are 3 counters but 10 people waiting to pay their bills so 10 people will be in 3 queues and who ever have completed paying the bills will leave the line and make way for next person.