TopolearnSeeing a message go through the broker in a dozen lines makes the model concrete.
pika is the standard Python client for AMQP 0-9-1 (pip install pika). Its BlockingConnection is the simplest API.
The steps are always the same:
- Open a connection and a channel.
- Declare the queue with
queue_declare. Declaring is idempotent: it creates the queue if missing and does nothing if it exists with the same settings. Both producer and consumer usually declare it, so start-up order doesn't matter. Declaring with different settings raises aPRECONDITION_FAILEDerror. - Publish with
basic_publish(exchange="", routing_key="hello", body=...). The empty string is the default exchange: every queue is automatically bound to it by its own name, so the routing key acts as the queue name. - Consume with
basic_consume(queue, on_message_callback)andstart_consuming(), which blocks and calls your function for each message.basic_getfetches a single message, handy in scripts and tests but inefficient for real consumers.
The body is bytes. Choose a serialization (usually JSON) and set content_type in the message properties so consumers know how to decode it.
Here auto_ack=True means the broker forgets a message as soon as it's sent; the next lessons show why real consumers acknowledge manually.
Example
# pip install pika
import json, pika
conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
ch = conn.channel()
ch.queue_declare(queue="hello") # idempotent
for i in range(3):
ch.basic_publish(exchange="", routing_key="hello", # default exchange
body=json.dumps({"n": i}),
properties=pika.BasicProperties(content_type="application/json"))
def on_message(channel, method, properties, body):
print("got", json.loads(body))
if json.loads(body)["n"] == 2:
channel.stop_consuming()
ch.basic_consume(queue="hello", on_message_callback=on_message, auto_ack=True)
ch.start_consuming() # got {'n': 0} got {'n': 1} got {'n': 2}
conn.close()Remember
- Declare queues idempotently from both sides
- exchange="" is the default exchange; routing_key = queue name
- Bodies are bytes; set content_type and serialize explicitly
Watch out for
- Declaring an existing queue with different arguments (e.g. durable) and getting PRECONDITION_FAILED.