CometMQAppearance
AMQP Basics
Essential AMQP 0.9.1 concepts: connections, channels, exchanges, queues, bindings, reliability, and durability.
Protocol Overview
AMQP 0.9.1 is a binary protocol with multiplexed channels, publisher confirms, and consumer acknowledgments.
Entities
Connection
python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
Channel
python
channel = connection.channel()
Virtual Host
Namespace isolation. Default is /.
Message
Properties plus payload.
Message Flow
Publisher → Exchange → Bindings → Queue → Consumer
Reliability
Publisher Confirms
python
channel.confirm_delivery()
channel.basic_publish(exchange='', routing_key='my-q', body='hello')
Consumer Acks
python
def cb(ch, method, props, body):
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume('my-q', cb, auto_ack=False)
channel.start_consuming()
Persistent Messages
Use delivery_mode=2 together with a durable queue.
QoS Prefetch
python
channel.basic_qos(prefetch_count=10)
Exchanges
Types: direct, fanout, topic, headers.
python
channel.exchange_declare('logs.direct', 'direct', durable=True)
channel.exchange_declare('notifications', 'fanout', durable=True)
channel.exchange_declare('logs.topic', 'topic', durable=True)
Topic wildcards: * is one word, # is zero or more. Default exchange: empty string ''.
Queues
python
channel.queue_declare('durable-q', durable=True)
channel.queue_declare('exclusive-temp', exclusive=True)
channel.queue_declare('auto-clean', auto_delete=True)
Persistent publish:
python
channel.basic_publish('', 'durable-q', 'Important', properties=pika.BasicProperties(delivery_mode=2))
TTL / DLX:
python
channel.queue_declare('main', durable=True, arguments={
'x-message-ttl': 30000,
'x-dead-letter-exchange': 'dlx',
'x-dead-letter-routing-key': 'dead.main'
})
Max length & overflow:
python
channel.queue_declare('bounded', durable=True, arguments={
'x-max-length': 5000,
'x-overflow': 'reject-publish-dlx',
'x-dead-letter-exchange': 'overflow-dlx'
})
Bindings
python
channel.queue_bind('error-q', 'logs.direct', 'error')
channel.queue_bind('all', 'logs.topic', '#')
channel.queue_bind('errors', 'logs.topic', '*.error.*')
Reliability & Flow Control
Memory and disk alarms block publishes under pressure. Publisher callbacks (Python):
python
def on_connection_blocked(conn, reason): pass
def on_connection_unblocked(conn): pass
connection.add_on_connection_blocked_callback(on_connection_blocked)
connection.add_on_connection_unblocked_callback(on_connection_unblocked)
Tactics: scale consumers, use TTL, reduce publish rate, add resources.
Durability & Persistence
Three layers:
- Durable exchanges — definitions survive restart
- Durable queues — definitions survive restart
- Persistent messages (
delivery_mode=2) — written to disk
Full resilience requires all three. WAL is a segmented journal; compaction reclaims space.
RPC Pattern (Direct Reply-To)
CometMQ supports the Direct Reply-To pattern for efficient RPC calls.
Traditional RPC Pattern
Client creates a temporary exclusive callback queue for each request:
python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# Create temporary callback queue
callback_queue = channel.queue_declare('', exclusive=True).method.queue
# Publish request with reply_to
channel.basic_publish(
exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(reply_to=callback_queue, correlation_id='123'),
body='request'
)
# Consume response
for method, props, body in channel.consume(callback_queue):
if props.correlation_id == '123':
print(f"Response: {body}")
break
Drawbacks: overhead of creating/deleting temporary queues for each RPC session.
Direct Reply-To Pattern (Optimized)
Client subscribes to a special pseudo-queue and messages are delivered directly to the channel without creating a queue:
python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# Subscribe to direct-reply-to pseudo-queue
channel.basic_consume('amq.rabbitmq.reply-to', lambda ch, method, props, body: print(body))
# Publish request with direct reply-to
channel.basic_publish(
exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(reply_to='amq.rabbitmq.reply-to', correlation_id='123'),
body='request'
)
channel.start_consuming()
Benefits:
- No queue creation/deletion overhead
- Lower latency and resource usage
- Direct message delivery to channel
- Better performance under high RPC load
Using aio-pika-msgpack-rpc
High-level RPC with automatic direct-reply-to:
python
import asyncio
import aio_pika
from aio_pika_msgpack_rpc import MSGPackRPC
async def server():
connection = await aio_pika.connect_robust('amqp://localhost/')
channel = await connection.channel()
rpc = await MSGPackRPC.create(channel)
async def add(a, b):
return a + b
await rpc.register("add", add)
# Server keeps running...
async def client():
connection = await aio_pika.connect_robust('amqp://localhost/')
channel = await connection.channel()
rpc = await MSGPackRPC.create(channel)
result = await rpc.call("add", kwargs={"a": 2, "b": 3})
print(f"Result: {result}") # Output: 5
How Direct Reply-To Works
- Client subscribes to direct-reply-to pseudo-queue → broker assigns unique consumer tag
- Client publishes request with direct reply-to address in
reply_toproperty - Server receives request and publishes response to the reply-to address
- Broker routes response directly to client's channel (no queue involved)
- Client receives response via its consumer
The broker automatically handles routing based on correlation_id to match responses to requests.
Performance Considerations
- Traditional RPC: ~2-3 ms overhead per request (queue create/delete)
- Direct Reply-To: ~0.1 ms overhead (direct routing)
- Concurrent requests: correlation_id ensures correct response matching
- Scalability: thousands of concurrent RPC sessions without queue overhead
Best Practices
- Use durable + persistent for critical data
- TTL + DLX for retries and cleanup
- Consistent routing key taxonomy
- Monitor queue depth and limits
- Avoid unnecessary bindings
- Use direct-reply-to for high-performance RPC