Соединение, канал и долговечная топология

У события есть договор, но ещё нет места доставки. Теперь создадим соединение с RabbitMQ, откроем канал и объявим exchange с очередями. Результат урока — повторяемая топология проекта «Доставка», где каталог и аудит имеют собственные долговечные очереди. Отдельно разберём, что означает долговечность в одноузловой учебной среде.

Полные функции connect и declare находятся в broker.py. Они не выполняются при импорте модуля. Сервер и клиент не запускались; читатель готовит отдельное окружение по README. Официальный первый Python-урок RabbitMQ подтверждает базовое использование Pika, а руководство по соединениям описывает их жизненный цикл.

Соединение переносит несколько каналов

Connection представляет сетевую связь приложения с брокером. Channel является логическим потоком операций внутри неё. Объявление очереди, публикация, получение и ack выполняются через канал. Это позволяет не открывать отдельное TCP-соединение для каждого короткого действия. Но при отказе connection связанные каналы тоже больше не пригодны для продолжения работы.

В учебном примере используем BlockingConnection: код последовательно выполняет операции и проще читается на первых шагах. Такой адаптер не означает, что любая длительная обработка безопасна внутри callback. Обработка входящих кадров и heartbeat зависит от того, получает ли клиент возможность обслуживать соединение. Позже учтём эту границу при обсуждении восстановления.

Полная функция подключения проекта:

def connect():
    parameters = pika.URLParameters(os.environ.get(
        'AMQP_URL', 'amqp://guest:guest@localhost:5672/%2F'
    ))
    parameters.heartbeat = 60
    parameters.blocked_connection_timeout = 30
    parameters.socket_timeout = 10
    return pika.BlockingConnection(parameters)

В URI указан virtual host /, закодированный как %2F. Имя virtual host относится к логическому пространству объектов брокера, а не к папке Python-проекта. Для независимых учебных опытов можно использовать отдельное пространство и отдельного пользователя, если это согласовано с настройками выбранного RabbitMQ.

Значения времени выбраны как конкретная учебная конфигурация, не как измеренный оптимум. Heartbeat помогает обнаруживать неработающую связь, socket timeout ограничивает определённые сетевые ожидания, blocked timeout связан с блокировкой соединения брокером. Они имеют разные назначения; один параметр не заменяет все возможные дедлайны приложения.

Объекты объявляются с известными свойствами

В declare сначала объявляем topic-exchange delivery.events. Затем direct-exchange delivery.resume для возвращения отложенной попытки конкретному обработчику. Для каждого имени catalog и audit создаём основную очередь, очередь retry и отдельную dead-очередь. Поздние объекты присутствуют в итоговом архиве заранее, но их механизм изучим после подтверждений.

Для первых сообщений существенна следующая часть полного declare:

channel.exchange_declare(
    exchange='delivery.events', exchange_type='topic', durable=True
)
for name in ('catalog', 'audit'):
    channel.queue_declare(
        queue=f'delivery.{name}', durable=True,
        arguments={'x-queue-type': 'classic'}
    )
    channel.queue_bind(
        queue=f'delivery.{name}', exchange='delivery.events',
        routing_key='material.*'
    )

Повторное объявление существующего объекта с совместимыми свойствами подтверждает его наличие. Но это не универсальная миграция. Если очередь с тем же именем уже имеет другой тип или несовместимые аргументы, брокер может закрыть канал с ошибкой. Нельзя воспринимать любой отказ объявления как временную сеть и бесконечно повторять неверную конфигурацию.

Имена фиксированы в коде потому, что получатели должны знать, куда возвращаться после перезапуска. Случайная exclusive-очередь полезна для временного наблюдателя, но исчезает с жизненным циклом соединения и не подходит накоплению важной работы при остановленном обработчике. Для каталога сохраняем отдельную известную очередь.

Durable описывает объект, persistent — сообщение

Флаг durable относится к exchange или queue. Для важного сообщения дополнительно зададим persistent delivery mode. Это два разных уровня: долговечная очередь может получить недолговечное сообщение, а persistent-сообщение в недолговечной очереди не превращает очередь в сохраняемый объект. Следующий урок соединит эти параметры при публикации.

В проекте используются classic queues одного учебного узла. Они не дают репликации содержимого на другие узлы. Устойчивость к обычному перезапуску и устойчивость к потере диска или узла являются разными задачами. В заключительных уроках рассмотрим, зачем для рабочих сценариев изучать quorum queues и соответствующие условия отказа.

Не меняйте тип существующей очереди с classic на quorum простым повторным объявлением. Тип является свойством создания, и миграция требует отдельного плана объектов и сообщений. Даже если удалить учебную очередь и создать новую, её прежние сообщения будут потеряны. Поэтому такие действия не скрываются внутри «удобной» функции подключения.

Ожидаемое состояние после будущего объявления:

delivery.events: topic exchange
delivery.catalog: долговечная очередь каталога
delivery.audit: долговечная очередь аудита
binding material.*: отдельный маршрут к каждой очереди
connection/channel: живут в текущем процессе клиента

Последняя строка напоминает, что сохранённая топология не сохраняет объект Python-channel. После потери связи приложение создаёт новое соединение и новый канал, а затем снова подтверждает нужные объявления. Это восстановление клиента, а не восстановление самого брокера из резервной копии.

Кто вправе объявлять объекты

Локальный guest используется только как понятное исходное значение для самостоятельной среды. Он не является рекомендуемым пользователем удалённого сервиса. Для рабочего пространства права конфигурации, записи и чтения ограничиваются известными ресурсами. Издателю необязательно разрешать удаление всех очередей, а обработчику аудита — чтение очереди каталога.

В итоговом примере каждый процесс вызывает declare, чтобы читатель не зависел от скрытого предварительного скрипта. При рабочей эксплуатации топологией часто управляет отдельный механизм, а приложения проверяют нужный договор с более узкими правами. Такой переход должен быть описан явно: если убрать права configure, прежний startup-код может перестать работать.

URI подключения содержит секрет, поэтому не печатаем его целиком в журнал. Для диагностики достаточно имени процесса, типа отказа и безопасного идентификатора окружения. Секрет не должен попадать в Markdown-исходники, sample-event или ZIP. В README показываем имя переменной, а действительное значение читатель задаёт локально.

Топология появляется раньше первой важной публикации. Если создать exchange, отправить сообщение и только затем объявить очередь, отсутствующая очередь не получит прошлую работу задним числом. Durable тоже не исправляет этот порядок: оно относится к уже существующему объекту. Именно поэтому полный издатель вызывает declare до publish, а не рассчитывает, что consumer обязательно успел стартовать первым.

Закрытие соединения после команды не удаляет наши долговечные очереди. Сообщения могут ждать consumer, пока выбранное хранилище broker сохраняет их. Но Python-ссылки на connection и channel после закрытия больше не пригодны для следующей команды. При будущем повторе снова создаём клиентскую связь, сохраняя те же предметные имена объектов.

В некоторых окружениях правила topology задаются политиками broker. Жёсткие аргументы queue_declare в учебном коде делают пример самодостаточным, но затрудняют изменение части настроек без миграции объекта. Для рабочего сопровождения нужно определить, какие параметры принадлежат коду, а какие операторской политике. Два независимых источника конфигурации не должны молча противоречить друг другу.

Для упражнения представьте очередь delivery.catalog с другим типом под тем же именем. Объясните, почему повторное объявление не должно молча преобразовывать её. Затем сопоставьте два процесса одной очереди и две независимые очереди. Далее опубликуем известное событие и разберём свойства сообщения.

Оглавление серии