RabbitMQ non-blocking consumer

Question

I'm using RabbitMQ in Python to manage several queues between a producer and multiple consumers. In the example in RabbitMQ website (routing model), the consumers are blocked. It means that they stop on start_consuming() and execute the callback function every time there is a new "task" in the queue.

My question is: how can I implement my consumer in a way that he is still waiting for tasks (so, the callback function is called every time there is new things in the queue) but at the same time he can execute other work/code.

Thank you

Solution

Form FAQ:

Pika does not have any notion of threading in the code. If you want to use Pika with threading, make sure you have a Pika connection per thread, created in that thread. It is not safe to share one Pika connection across threads,

So lets create the connection inside the thread:

import pika


class PikaMassenger():

    exchange_name = '...'

    def __init__(self, *args, **kwargs):
        self.conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        self.channel = self.conn.channel()
        self.channel.exchange_declare(
            exchange=self.exchange_name, 
            exchange_type='topic')

    def consume(self, keys, callback):
        result = self.channel.queue_declare('', exclusive=True)
        queue_name = result.method.queue
        for key in keys:
            self.channel.queue_bind(
                exchange=self.exchange_name, 
                queue=queue_name, 
                routing_key=key)

        self.channel.basic_consume(
            queue=queue_name, 
            on_message_callback=callback, 
            auto_ack=True)

        self.channel.start_consuming()


    def __enter__(self):
        return self


    def __exit__(self, exc_type, exc_value, traceback):
        self.conn.close()

def start_consumer():

    def callback(ch, method, properties, body):
        print(" [x] %r:%r consumed" % (method.routing_key, body))

    with PikaMassenger() as consumer:
        consumer.consume(keys=[...], callback=callback)


consumer_thread = threading.Thread(target=start_consumer)
consumer_thread.start()

Comments

[loopbackbee]

why not run start_consuming() on a separate thread?

[Hugo Sousa]

Well, I got a solution pretty easy. Instead of using basic_consume, I can simple use the basic_get inside a function and call this function every X seconds. But there is a question: would the queue tasks be delivered with some order?

[Andriy Drozdyuk]

@HugoSousa Would be great if you could post a full solution here - I am new to rabbitmq and it would really help others.

[kwarunek]

another option is to use pika with Tornado fully async (pika.readthedocs.org/en/latest/examples/tornado_consumer.ht‌​ml)

[Incömplete]

@goncalopp according to FAQ, pika is not threadsafe

[adnanmuttaleb]

why this question is not answered yet!!

[Robben_Ford_Fan_boy]

Pika is not thread safe - the API does not support this functionality

[m.raynal]

Robben_Ford is right, one should not use threads along with pika. "Pika does not have any notion of threading in the code. If you want to use Pika with threading, make sure you have a Pika connection per thread, created in that thread. It is not safe to share one Pika connection across threads." -> From pika doc FAQ