Reputation: 31
I have two python processes, one consumer process and one producer process. Each process will start a rabbitmq connection and spawn multiple consumer/producer threads. Each thread will create a channel in the connection and perform the message sending and receiving logic.
This is my consumer thread
def consumer_thread(connection, routing_key):
channel = connection.channel()
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange="test", routing_key=routing_key, queue=queue_name)
thread_name = current_thread().name
def process(ch, method, properties, body):
print(f"{thread_name} received {body}")
channel.basic_consume(process, queue=queue_name, no_ack=True)
channel.start_consuming()
This is my producer thread
def producer_thread(connection, routing_key, sleep_time):
channel = connection.channel()
thread_name = current_thread().name
count = 0
while True:
count += 1
channel.basic_publish("test", routing_key=routing_key,
body=f"msg {count} from {thread_name}")
time.sleep(sleep_time)
And I start a rabbitmq connection using
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
However, when I run my code, for the first message received at consumer thread, I am getting this error message
Traceback (most recent call last):
File "D:\app\cortex-bin\Python36\lib\threading.py", line 916, in _bootstrap_inner
self.run()
File "D:\app\cortex-bin\Python36\lib\threading.py", line 864, in run
self._target(*self._args, **self._kwargs)
File "D:\app\cortex\background\core\scratch\test.py", line 18, in consumer_thread
channel.start_consuming()
File "D:\app\cortex-bin\Python36\lib\site-packages\pika\adapters\blocking_connection.py", line 1817, in start_consuming
'start_consuming may not be called from the scope of 'pika.exceptions.RecursionError: start_consuming may not be called from the scope of another BlockingConnection or BlockingChannel callback'
For all subsequent messages, they can be received by the consumer threads just fine.
May I know what's causing this exception? Thanks.
Upvotes: 0
Views: 1955
Reputation: 9667
You can't access a Pika connection from multiple threads (comment). Your threads must start their own connection and channels.
Upvotes: 1