Pika Python AMQP Client

Warnings

• breaking Pika 1.0.0 introduced significant breaking changes, especially around connection parameters and callback signatures. `ConnectionParameters` no longer accepts `credentials` directly; `pika.PlainCredentials` (or similar) must be used. Also, the `on_message_callback` signature changed.
  Fix: For `ConnectionParameters`, create a `pika.credentials.PlainCredentials` object and pass it via `pika.ConnectionParameters(credentials=...)`. Update `on_message_callback` to accept `(ch, method, properties, body)`.
• gotcha Mixing blocking (BlockingConnection) and non-blocking (AsyncioConnection, TornadoConnection) I/O models in the same application without proper isolation can lead to deadlocks, unexpected behavior, or resource exhaustion. Choose one I/O model and stick to it, or use separate processes for different models.
  Fix: Ensure you understand Pika's different I/O adapters. Use `BlockingConnection` for synchronous applications or scripts, and an appropriate `pika.adapters` class for asynchronous applications integrated with an event loop (e.g., `asyncio`, `Tornado`). Do not call blocking methods on a channel managed by an async adapter outside of its event loop.
• gotcha Failure to explicitly close channels and connections can lead to resource leaks (e.g., open file descriptors), hung processes, and issues with the RabbitMQ broker. Pika 1.0.0 and later require explicit closure for `BlockingConnection`.
  Fix: Always ensure that `channel.close()` and `connection.close()` are called, typically in a `finally` block, after your AMQP operations are complete or when an exception occurs. For asynchronous adapters, the `stop()` method often handles this gracefully.
• gotcha Pika does not automatically handle connection or channel recovery by default. Network outages, broker restarts, or protocol errors will cause `pika.exceptions.ConnectionClosed` or `pika.exceptions.ChannelClosed`.
  Fix: Implement robust error handling and reconnection logic in your application. For `BlockingConnection`, this typically involves a loop that attempts to re-establish the connection and channel after a delay. For asynchronous adapters, specific reconnection examples are often provided in the Pika documentation.

Install

• pip install pika Install stable version

Imports

• BlockingConnection

  from pika import BlockingConnection

  The BlockingConnection class is directly available under the pika namespace since version 1.0.0.
• ConnectionParameters

  from pika import ConnectionParameters

• PlainCredentials

  from pika import PlainCredentials

  PlainCredentials is available directly in the pika namespace since 1.0.0, and ConnectionParameters no longer accepts the credentials argument directly.
• AsyncioConnection

  from pika.adapters.asyncio_connection import AsyncioConnection

  Use the appropriate adapter for your asynchronous framework (e.g., asyncio, tornado).

Quickstart

This quickstart demonstrates a simple Pika producer using BlockingConnection to connect to a RabbitMQ instance (defaulting to localhost) and publish a message to a 'hello' queue. It also shows the basic setup for a consumer callback (commented out) and proper connection/channel closure.

import pika
import os

def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")

# Get RabbitMQ host from environment or default to localhost
rabbitmq_host = os.environ.get('RABBITMQ_HOST', 'localhost')

connection = None
channel = None
try:
    # Establish connection
    connection = pika.BlockingConnection(pika.ConnectionParameters(host=rabbitmq_host))
    channel = connection.channel()

    # Declare a queue
    channel.queue_declare(queue='hello')

    # Publish a message
    channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
    print(" [x] Sent 'Hello World!'")

    # Start consuming (example consumer setup)
    # channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
    # print(' [*] Waiting for messages. To exit press CTRL+C')
    # channel.start_consuming()

finally:
    # Ensure channel and connection are closed
    if channel and channel.is_open:
        channel.close()
    if connection and connection.is_open:
        connection.close()