Using RabbitMQ with AsyncIO

The defacto library to use with RabbitMQ with Python’s AsyncIO is the aioamqp library.

Issues with persistent connections

If you have long running processes built around listening to queues and publishing to queues, you need to make sure your connection to RabbitMQ stays open and stable.

RabbitMQ and heartbeat

RabbitMQ has a feature called heartbeat. The setting to tweak this is not quite what you’d expect. In fact, if you didn’t do a deep read into the documentation on what heartbeat is doing(or if you relied on the docs from libraries), you might think the setting meant how often a library would send heartbeats to RabbitMQ

Heartbeat setting explained

The heartbeat timeout value defines after what period of time the peer TCP connection should be considered unreachable (down) by RabbitMQ and client libraries. This value is negotiated between the client and RabbitMQ server at the time of connection.

Using heartbeats properly

If you want to do all you can to prevent connections from getting dropped by RabbitMQ, you’ll want to have a large number here.

import aioamqp
transport, protocol = await aioamqp.connect(
    host, port,
    'guest', 'guest',
    heartbeat=800
)

When are heartbeats issued

RabbitMQ will issue heartbeats to the server about every 2 seconds if there is activity; however, if there is no activity, no heartbeat frames will be sent and you could get disconnected.

However, there is a solution. You can manually send out heartbeats in an asyncio task.

import aioamqp
import asyncio


async def heartbeat(protocol):
    while True:
        await asyncio.sleep(20)  # issue manual heartbeat every 20 seconds
        await protocol.send_heartbeat()


transport, protocol = await aioamqp.connect(
    host, port,
    'guest', 'guest',
    heartbeat=800
)

asyncio.ensure_future(heartbeat(protocol))

Handling other errors

Even with the heartbeat feature, you can still experience issues with disconnections and other errors.

aioamqp has another feature however that allows you to do handle when any kind of disconnect occurs. It provides a wait_closed coroutine, which finishes when the connection to RabbitMQ is done.

So you can setup an asyncio task to wait for this and handle reconnects when it comes across it.

import aioamqp
import asyncio


async def handle_closed(protocol):
    await protocol.wait_closed()
    # reconnect logic here...


transport, protocol = await aioamqp.connect(
    host, port,
    'guest', 'guest',
    heartbeat=800
)

asyncio.ensure_future(handle_closed(protocol))