Each Answer to this Q is separated by one/two green lines.
I am trying to create a consumer that would subscribe to multiple queues, and then process messages as they arrive.
The problem is that when there is some data already present in the first queue, it consumes the first queue and never goes to consume the second queue.
However, when the first queue is empty, it does go to the next queue, and then consumes both queues simultaneously.
I had first implemented threading but want to steer clear of it, when pika library does it for me without much complexity. Below is my code:
import pika mq_connection = pika.BlockingConnection(pika.ConnectionParameters('x.x.x.x')) mq_channel = mq_connection.channel() mq_channel.basic_qos(prefetch_count=1) def callback(ch, method, properties, body): print body mq_channel.basic_ack(delivery_tag=method.delivery_tag) mq_channel.basic_consume(callback, queue="queue1", consumer_tag="ctag1.0") mq_channel.basic_consume(callback, queue="queue2", consumer_tag="ctag2.0") mq_channel.start_consuming()
One possible solution is to use non blocking connection and consume messages.
import pika def callback(channel, method, properties, body): print(body) channel.basic_ack(delivery_tag=method.delivery_tag) def on_open(connection): connection.channel(on_channel_open) def on_channel_open(channel): channel.basic_consume(callback, queue="queue1") channel.basic_consume(callback, queue="queue2") parameters = pika.URLParameters('amqp://guest:[email protected]:5672/%2F') connection = pika.SelectConnection(parameters=parameters, on_open_callback=on_open) try: connection.ioloop.start() except KeyboardInterrupt: connection.close()
This will connect to multiple queues and will consume messages accordingly.
The issue is most likely that the first call has issued a Basic.Consume and has already received messages from a pre-populated queue before the second call is issued. You might want to try setting the QoS prefetch count to 1, which will limit RabbitMQ from sending you more than one message at a time.
Similar to comments in the first answer above, I was able to get similar results with pika 1.1.0 and the following:
import pika def queue1_callback(ch, method, properties, body): print(" [x] Received queue 1: %r" % body) def queue2_callback(ch, method, properties, body): print(" [x] Received queue 2: %r" % body) def on_open(connection): connection.channel(on_open_callback = on_channel_open) def on_channel_open(channel): channel.basic_consume('queue1', queue1_callback, auto_ack = True) channel.basic_consume('queue2', queue2_callback, auto_ack = True) credentials = pika.PlainCredentials('u', 'p') parameters = pika.ConnectionParameters('localhost', 5672, "https://stackoverflow.com/", credentials) connection = pika.SelectConnection(parameters = parameters, on_open_callback = on_open) Try: connection.ioloop.start() except KeyboardInterrupt: connection.close() connection.ioloop.start()