Using the BlockingChannel.consume generator to consume messages

The BlockingChannel.consume method is a generator that will return a tuple of method, properties and body.

When you escape out of the loop, be sure to call BlockingChannel.cancel to requeue any messages the generator is still holding.

Note

cancel returns 0 and nothing else. The count has been vestigial since 0.10.0: only messages consumed with auto_ack=False can be requeued, and any arriving after the Basic.Cancel are discarded unacknowledged rather than counted, so the number is not knowable. Do not read the return value as a requeue count.

Example of consuming messages and acknowledging them:

import pika

connection = pika.BlockingConnection()
channel = connection.channel()

# Declare the queue we are about to consume from, so the example is
# self-contained. `durable=True` is required by RabbitMQ 4.3; see
# examples/README.md.
channel.queue_declare(queue='test', durable=True)

# Publish the ten messages the loop below expects, so it terminates.
for number in range(10):
    channel.basic_publish(exchange='',
                          routing_key='test',
                          body=f'Hello World {number}!'.encode())

# Get ten messages and break out
for method_frame, properties, body in channel.consume('test'):

    # Display the message parts
    print(method_frame)
    print(properties)
    print(body)

    # Acknowledge the message
    channel.basic_ack(method_frame.delivery_tag)

    # Escape out of the loop after 10 messages
    if method_frame.delivery_tag == 10:
        break

# Cancel the consumer and return any pending messages
requeued_messages = channel.cancel()
print(f'Requeued {requeued_messages} messages')

# Close the channel and the connection
channel.close()
connection.close()

The example publishes the ten messages it then consumes, so it terminates on its own. Its output on stdout looks like this, abbreviated to the first and last message. The consumer tag is a fresh UUID each run, and BlockingConnection also writes a deprecation warning to stderr:

$ python examples/blocking_consumer_generator.py
<Basic.Deliver(['consumer_tag=ctag1.5530b02c97654528aacbde82ba5cc271', 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=test'])>
<BasicProperties>
b'Hello World 0!'
...
<Basic.Deliver(['consumer_tag=ctag1.5530b02c97654528aacbde82ba5cc271', 'delivery_tag=10', 'exchange=', 'redelivered=False', 'routing_key=test'])>
<BasicProperties>
b'Hello World 9!'
Requeued 0 messages

If the queue already held messages, they are delivered first and take the low delivery tags, so the example's own ten arrive after them and the loop breaks before reaching all of them. Delivery tags are per channel and always start at 1.