[Python] How to use multithreading with Pika and RabbitMQ to perform requests and responses...

I'm working on a project with RabbitMQ. I'm using the RPC Pattern, basically I'm receiving or consuming messages from a queue, make some processing and then send a response back. I'm using Pika, my goal is to use a thread per task so for every Task I'll make a thread particularly for that task. I also read that the best practice is to make only one connection and under it many channels as I want to, but I get always this error:

pika.exceptions.RecursionError: start_consuming may not be called from the scope of another BlockingConnection or BlockingChannel callback.
​

I made some research and found out that Pika is not thread safe and we should use for every thread an independent connection and a channel. But I don't want to do that since it is considered bad practice. So I wanted to ask here if someone already achieved to make this work.

I read also that it is possible if I didn't use BlockingConnection to instantiate my connection and also that there is a function called add_callback_threadsafe which can make this possible. but there is unfortunately no examples for that and I read the documentation but it's complex and without examples it was hard for me to grasp what they want to describe.

My attempt was to declare two classes. Each class will represent a task Executer which receive or consume a message from a queue and based on that made some processing and deliver a response back. my idea was to share a RabbitMQ connection between the two tasks but every task will get an independent channel.

In the code above the rabbit Parameter passed to the function is a class that holds some variables like connection and other functions like EventSubscriber which when called it will assign a new Channel and start consuming messages from that Particular Exchanges and routingKey. Next I declare a Thread and give the subscribe or consume function as a Target to that thread. The other task class look also the same as this Class that's why I'll only upload this code. In the main class I make a connection to RabbitMQ and pass it as parameter to the constructor of the two task classes.

class On_Deregistration:

def __init__(self, rabbit):
self.event(rabbit) # this will call event function and pass the connection shared between all Tasks. rabbit parameter hold a connection to rabbitmq

def event(self, rabbit):

self.Subscriber = rabbit.EventSubscriber(rabbit, 'testing.test', 'test', False, onDeregistrationFromHRS # this func is task listener)

def subscribeAsync(self):
self.Subscriber.subscribe() # here i call start_consuming

def start(self):
"""start Subscribtion in an Independant Thread """
thread = threading.Thread(target = self.subscribeAsync )
thread.start()
if thread.isAlive():
print("asynchronous subscription started")


# Main Class:
class App:

def __init__(self):

self.rabbitMq = RabbitMqCommunicationInterface(host='localhost', port=5672)
firstTask = On_Deregistration(self.rabbitMq)
secondTask = secondTask(self.rabbitMq)

app = App()


I searched for the cause of this error and obviously Pika is not thread safe but there must be a solution for this. Maybe not using a BlockingConnection ? Maybe I'm missing something about how to implement multithreading with RabbitMQ.

Continue reading...