pika

Introduction

Pika is a pure-Python implementation of the AMQP 0-9-1 protocol including RabbitMQ's extensions.

• Supports Python 3.7+ (CPython and PyPy; versions are listed on PyPI). Pika 1.1.0 was the last release that supported Python 2.7.
• Since threads aren't appropriate to every situation, it doesn't require threads. Pika core takes care not to forbid them, either. The same goes for greenlets, callbacks, continuations, and generators. Most connection adapters are single-threaded; use Connection when you need to publish or consume from multiple threads.
• People may be using direct sockets, plain old select(), or any of the wide variety of ways of getting network events to and from a Python application. Pika tries to stay compatible with all of these, and to make adapting it to a new environment as simple as possible.

Documentation

Pika's documentation can be found here.

Example

Sending a message with pika.Connection, which is thread-safe and runs its own IOLoop on a background thread:

import pika

connection = pika.Connection(pika.ConnectionParameters())
channel = connection.channel()

# RabbitMQ 4.3 and later refuse a queue that is neither durable nor exclusive.
channel.queue_declare('test', durable=True)
channel.basic_publish(exchange='', routing_key='test', body=b'Test message.')
connection.close()

And consuming, where each delivery is dispatched on a per-channel worker thread, so slow processing does not stall heartbeats and basic_ack needs no add_callback_threadsafe:

import threading

import pika

connection = pika.Connection(pika.ConnectionParameters())
channel = connection.channel()
channel.queue_declare('test', durable=True)

def on_message(ch, method_frame, properties, body):
    print(method_frame, properties, body)
    ch.basic_ack(method_frame.delivery_tag)

channel.basic_consume('test', on_message)

try:
    # The IOLoop already runs in the background; block until interrupted.
    threading.Event().wait()
except KeyboardInterrupt:
    connection.close()

Adapters

Pika provides the following adapters:

• pika.Connection - thread-safe, runs SelectConnection's IOLoop in a background thread. All channel methods are safe to call from any thread simultaneously.
• pika.SelectConnection - asynchronous adapter without third-party dependencies.
• pika.adapters.asyncio_connection.AsyncioConnection - asynchronous adapter for AsyncIO's I/O loop.
• pika.BlockingConnection - synchronous and procedural (deprecated, removed in Pika 2.0).
• pika.adapters.gevent_connection.GeventConnection - asynchronous adapter for use with Gevent's I/O loop (deprecated, removed in Pika 2.0).
• pika.adapters.tornado_connection.TornadoConnection - asynchronous adapter for use with Tornado's I/O loop (deprecated, removed in Pika 2.0).
• pika.adapters.twisted_connection.TwistedProtocolConnection - asynchronous adapter for use with Twisted's I/O loop (deprecated, removed in Pika 2.0).

Multiple connection parameters

You can also pass multiple pika.ConnectionParameters instances for fault-tolerance as in the code snippet below (host names are just examples, of course). To enable retries, set connection_attempts and retry_delay as needed in the last pika.ConnectionParameters element of the sequence. Retries occur after connection attempts using all of the given connection parameters fail.

import pika

parameters = (
    pika.ConnectionParameters(host='rabbitmq.zone1.yourdomain.com'),
    pika.ConnectionParameters(host='rabbitmq.zone2.yourdomain.com',
                              connection_attempts=5, retry_delay=1))
connection = pika.BlockingConnection(parameters)

With non-blocking adapters, such as pika.SelectConnection and pika.adapters.asyncio_connection.AsyncioConnection, you can request a connection using multiple connection parameter instances via the connection adapter's create_connection() class method.

Threading

Using Connection (recommended)

pika.Connection is the simplest way to use Pika from multiple threads. It runs the IOLoop in a background thread and provides a blocking API that is safe to call from any number of threads simultaneously. Consumer callbacks run on a per-channel worker thread, so slow processing does not stall heartbeats as long as the callback keeps pace with deliveries:

import pika

conn = pika.Connection(pika.ConnectionParameters('localhost'))
ch = conn.channel()

# RabbitMQ 4.3 and later refuse a queue that is neither durable nor exclusive.
ch.queue_declare('work', durable=True)

def on_message(channel, method, properties, body):
    process(body)
    channel.basic_ack(method.delivery_tag)  # safe, no callback scheduling needed

ch.basic_consume('work', on_message)

See examples/basic_consumer_threaded.py and examples/basic_publisher_threaded.py.

Using add_callback_threadsafe (manual approach)

For other adapters, each connection instance is confined to a single thread. The single-threaded constraint may result in a dropped connection due to AMQP heartbeat timeout in consumers that take a long time to process a message. A common solution is to delegate processing to another thread while the connection's IOLoop thread services heartbeats.

Messages processed in another thread may not be acknowledged directly from that thread. Instead, schedule a callback in the adapter's IOLoop thread:

This is the beginning of the README. Read the rest at the original. Read the full README ↗