Установка и первое сообщение

Поднимаем RabbitMQ в одном контейнере, заглядываем в веб-панель и пишем пару producer + consumer, которая реально работает.

pika — основная Python-библиотека для работы с RabbitMQ по протоколу AMQP 0-9-1. Она не входит в стандартную библиотеку, поэтому примеры из этого урока запускаются на вашей машине, а не в браузере.

Запуск в Docker

RabbitMQ написан на Erlang, и ставить его руками из пакетов — удовольствие ниже среднего. Для разработки берите официальный образ с суффиксом -management: в нём сразу включена веб-панель.

docker run -d --name rabbit -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USER=dev -e RABBITMQ_DEFAULT_PASS=dev rabbitmq:3.13-management

Два порта — это не опечатка, и их постоянно путают:

5672 — AMQP: сюда подключаются ваши программы. Именно этот порт указывается в коде.
15672 — HTTP: веб-панель и management API. Сюда ходит браузер, но не pika.

Проверим, что брокер поднялся:

docker logs rabbit --tail 5
docker exec rabbit rabbitmq-diagnostics -q ping

Результат:

Ping succeeded

Если проект уже живёт в docker-compose, добавьте сервис туда — так вы не потеряете настройки после docker rm:

services:
  rabbit:
    image: rabbitmq:3.13-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: dev
      RABBITMQ_DEFAULT_PASS: dev
    volumes:
      - rabbit_data:/var/lib/rabbitmq   # очереди переживут пересоздание контейнера

volumes:
  rabbit_data:

Панель управления

Откройте http://localhost:15672 и войдите как dev / dev. Это не украшение, а главный инструмент отладки на первых порах:

Overview — общий график: сколько сообщений влетает и вылетает в секунду. Если линия входа выше линии выхода, вы копите отставание.
Queues and Streams — список очередей. Колонка Ready — сообщения ждут потребителя, Unacked — отданы, но ещё не подтверждены. Растёт Unacked — воркеры зависли на обработке.
Exchanges — обменники и привязки; отсюда же можно опубликовать тестовое сообщение руками, не написав ни строчки кода.
Connections / Channels — кто сейчас подключён. Пустой список при работающем приложении — верный признак, что вы стучитесь не туда.

Привыкайте держать вкладку с панелью открытой: половина «мистических» багов очереди видна на её графиках за пять секунд.

Первое сообщение

Ставим библиотеку:

pip install pika

Producer: отправляем

# send.py — кладём одно сообщение в очередь hello
import pika

credentials = pika.PlainCredentials("dev", "dev")
params = pika.ConnectionParameters(host="localhost", port=5672, credentials=credentials)

connection = pika.BlockingConnection(params)
channel = connection.channel()

channel.queue_declare(queue="hello", durable=True)

channel.basic_publish(
    exchange="",                 # обменник по умолчанию
    routing_key="hello",         # ...для него routing_key — это просто имя очереди
    body="Привет, RabbitMQ!".encode("utf-8"),
    properties=pika.BasicProperties(delivery_mode=2),  # persistent: писать на диск
)

print(" [x] Отправлено: Привет, RabbitMQ!")
connection.close()

Результат:

$ python send.py
 [x] Отправлено: Привет, RabbitMQ!

Загляните в панель: в разделе Queues появилась очередь hello, и в колонке Ready стоит единица. Сообщение лежит и ждёт. Можно закрыть ноутбук и вернуться завтра — оно никуда не денется.

Consumer: получаем

# receive.py — читаем сообщения из очереди hello, пока не нажмём Ctrl+C
import pika

credentials = pika.PlainCredentials("dev", "dev")
params = pika.ConnectionParameters(host="localhost", port=5672, credentials=credentials)

connection = pika.BlockingConnection(params)
channel = connection.channel()

channel.queue_declare(queue="hello", durable=True)
channel.basic_qos(prefetch_count=1)   # не давать воркеру больше одной задачи разом

def on_message(ch, method, properties, body):
    print(" [x] Получено:", body.decode("utf-8"))
    # здесь была бы настоящая работа: отправка письма, генерация отчёта...
    ch.basic_ack(delivery_tag=method.delivery_tag)   # подтверждаем: можно удалять

channel.basic_consume(queue="hello", on_message_callback=on_message)

print(" [*] Жду сообщений. Выход — Ctrl+C")
channel.start_consuming()

Результат:

$ python receive.py
 [*] Жду сообщений. Выход — Ctrl+C
 [x] Получено: Привет, RabbitMQ!

Consumer не завершается — он висит и ждёт следующих сообщений. Запустите send.py в соседнем терминале ещё пару раз: строки будут появляться мгновенно. Это и есть боевой режим воркера.

Построчный разбор

Строка	Что она делает и зачем
`BlockingConnection(params)`	TCP-соединение с брокером. Дорогая штука — открывается один раз на процесс, а не на каждое сообщение.
`connection.channel()`	Канал — лёгкая «виртуальная сессия» внутри соединения. Вся работа с AMQP идёт через каналы, а не напрямую через соединение.
`queue_declare(durable=True)`	Создаёт очередь, если её нет (идемпотентно). `durable` — очередь переживёт перезапуск брокера. Объявляют её обе стороны: кто первым запустился, тот и создал.
`exchange=""`	Обменник по умолчанию. У него особое свойство: он кладёт сообщение в очередь, чьё имя совпадает с `routing_key`. Отсюда иллюзия «шлём прямо в очередь».
`delivery_mode=2`	Persistent-сообщение: брокер запишет его на диск. Без этого сообщения живут только в памяти и исчезнут при рестарте RabbitMQ — даже из durable-очереди.
`basic_qos(prefetch_count=1)`	«Не давай мне следующее, пока я не подтвердил текущее». Без этого RabbitMQ веером раскидает всю очередь по воркерам, и один медленный заберёт половину задач впустую.
`basic_ack(delivery_tag=...)`	Подтверждение обработки. Только после него сообщение удаляется из очереди.

Как это работает

Уберите basic_ack из on_message и остановите consumer по Ctrl+C — сообщение вернётся в очередь и снова станет Ready. В этом весь смысл подтверждений: RabbitMQ считает сообщение обработанным не тогда, когда отдал его по сети, а тогда, когда воркер об этом сказал. Умер воркер, оборвалась сеть, процесс убили по OOM — сообщение уйдёт другому.

Отсюда два состояния в панели. Ready — лежит в очереди, ждёт свободного потребителя. Unacked — отдано воркеру, тот работает, брокер ждёт ack и держит копию у себя. Если Unacked висит и не уменьшается, значит воркер завис внутри обработки (например, в бесконечном ожидании внешнего API), и это первое место, куда стоит смотреть при разборе инцидента.

Ещё одна деталь, о которую спотыкаются все: для надёжности нужны обе настройки. durable=True сохраняет саму очередь, а delivery_mode=2 — сообщения внутри неё. Забудете второе — после перезапуска брокера очередь будет на месте, но пустая.

Частые ошибки

Подключаться на порт 15672. Это HTTP-панель. pika ждёт AMQP на 5672 и на 15672 даст непонятную ошибку соединения.
Логиниться как guest с другой машины. Встроенный пользователь guest по умолчанию пускают только с localhost. Из контейнера или с соседнего хоста вы получите ACCESS_REFUSED. Поэтому мы сразу завели dev.
Объявить очередь с разными параметрами. Producer объявил hello как durable=True, а consumer — как durable=False. Получите PRECONDITION_FAILED - inequivalent arg 'durable' и закрытый канал. Параметры queue_declare должны совпадать везде — держите их в общей константе.
Открывать соединение на каждое сообщение. TCP-хендшейк плюс AMQP-хендшейк — это десятки миллисекунд. В цикле на тысячу сообщений вы получите минуту вместо секунды. Соединение — один раз, канал — надолго.
Долгая обработка внутри callback. BlockingConnection однопоточный: пока вы пять минут генерируете отчёт, pika не отвечает на heartbeat, и брокер рвёт соединение с missed heartbeats. Ack потом не проходит, сообщение возвращается в очередь и обрабатывается заново — по кругу. Лечится увеличением heartbeat или вынесением работы в отдельный поток с add_callback_threadsafe.
Забыть, что body — это байты. Передали str — получите TypeError. Всегда .encode("utf-8") на отправке и .decode("utf-8") на приёме; для структур — json.dumps(...).encode("utf-8").

Итоги

Для разработки RabbitMQ поднимается одной командой: образ rabbitmq:3.13-management, порты 5672 (AMQP, для кода) и 15672 (веб-панель, для вас).
Панель на 15672 — главный отладчик: колонки Ready и Unacked честно рассказывают, что происходит с очередью.
Минимальный producer: соединение → канал → queue_declare → basic_publish. Минимальный consumer: то же плюс basic_consume и basic_ack в конце обработки.
Пустой exchange="" — это обменник по умолчанию, который маршрутизирует по имени очереди; настоящие обменники начнутся в следующем разделе.
Надёжность = durable=True у очереди + delivery_mode=2 у сообщения + честный basic_ack после обработки. Пропустите любой из трёх пунктов — потеряете сообщения.