Восстановление соединения и порядок редакций

У consumer есть квитанция и защита от старых редакций. Теперь вернёмся к жизненному циклу клиента: connection может оборваться, channel закрыться, а неподтверждённая работа вернуться в очередь. Результат урока — управляемое восстановление с паузой и понимание того, почему транспортный порядок не заменяет предметную последовательность revision.

Используем цикл запуска consumer.py и настройки connect в broker.py. Автор не останавливал узел и не выполнял reconnect. Официальное руководство по надёжности объясняет восстановление соединений и возможные повторы, а руководство по quorum queues задаёт отдельные свойства реплицируемого хранения.

Закрытый канал нельзя продолжить новой командой

После потери connection связанные с ним каналы больше не представляют живую связь. Нужно создать новый connection, новый channel, восстановить объявления и подписку. Нельзя просто повторить basic_ack старого delivery_tag через новый канал. Tag относится к прежней транспортной доставке, которую broker будет рассматривать по состоянию старой связи.

Функция consume создаёт соединение и канал на каждом вызове, включает confirms, задаёт prefetch и регистрирует consumer. В finally закрывает connection, если оно ещё открыто. Внешний цикл при известных AMQP, сетевых и SQLite-ошибках делает паузу, затем снова вызывает consume. Так восстановление не зависит от скрытого автоматического поведения клиента.

Фрагмент полного цикла запуска:

delay = 1
while True:
    try:
        consume(args.name)
        delay = 1
    except KeyboardInterrupt:
        break
    except (pika.exceptions.AMQPError, OSError, sqlite3.Error) as error:
        print('reconnect:', type(error).__name__, flush=True)
        time.sleep(delay)
        delay = min(delay * 2, 30)

Этот код не выполнялся автором. Пауза начинает расти после последовательных отказов и ограничивается тридцатью секундами. Для большого числа клиентов понадобятся jitter и более подробное различение ошибок. Иначе одновременно восстановленные процессы начнут повторять подключения синхронно. Учебная реализация показывает причину паузы, не заявляя оптимальные параметры.

Ошибка конфигурации не лечится reconnect

Несовместимое объявление существующей очереди или отсутствие права чтения может закрыть channel. Новый connection с прежней неверной конфигурацией воспроизведёт отказ. Наш цикл сохраняет работоспособность процесса для наблюдения, но сопровождающий должен отличить устойчивую ошибку договора от временной потери сети. Бесконечное ожидание не является исправлением неправильной topology.

В журнале печатается класс исключения, без URI и тела события. Для рабочего сопровождения этого может быть недостаточно: потребуется безопасное поле причины и контекст resource. Однако добавить полный traceback с секретами вместо предметного контекста тоже не решение. Диагностическая запись должна вести к конкретному объекту и действию.

Если consumer был отменён из-за удаления очереди, дальнейшая политика также должна быть осознанной. Самопроизвольное пересоздание всех объектов может скрыть чужое административное действие. В нашем отдельном учебном пространстве declare имеет заранее известные имена, поэтому восстановление остаётся ограниченным. Для рабочего broker управление ресурсами следует согласовать с оператором.

Heartbeat зависит от работы клиента

Heartbeat помогает обнаружить недоступное соединение и поддерживать активность для некоторых сетевых устройств. Но BlockingConnection должен иметь возможность обрабатывать служебные кадры. Если callback занят долгой синхронной операцией и не возвращается, выбранное время heartbeat не превращает такую работу в безопасную фоновую задачу.

Наш callback выполняет короткую локальную транзакцию и небольшую подтверждённую публикацию retry/dead. Это ограничение архитектуры. Для минутной обработки видео или больших внешних запросов потребуется другое управление выполнением и подтверждениями, например отдельный рабочий поток с безопасной передачей действий обратно в connection-поток либо асинхронный клиент.

Не переносите channel произвольно между потоками только потому, что объект Python доступен по ссылке. Правила потокобезопасности принадлежат выбранному адаптеру Pika. В серии нет параллельных callback внутри одного соединения, поэтому не возникает скрытого требования синхронизировать все операции channel. Изменение модели исполнения нужно проектировать отдельно.

Очерёдность доставки и порядок эффекта

Предположим, published revision 1 попало в retry, а withdrawn revision 2 прошло успешно. Через задержку старый факт возвращается в catalog. Если считать очередь гарантированным порядком предметной истории, публикация снова станет видимой. Но наша projection сравнивает revision и отмечает старое событие как stale. Это решение работает независимо от причин транспортной перестановки.

Даже без retry несколько consumer одной очереди могут завершать работу в другом порядке. Первый получил тяжёлую операцию, второй — короткую следующую. Prefetch ограничивает объём незавершённой работы, но не делает их commits единым последовательным процессом. Поэтому добавление второго consumer требует проверки предметного договора, а не только размера очереди.

Для строгой обработки всех изменений последовательно можно применять другие схемы: один исполнитель, разбиение по ключу материала или журнал с контролируемыми позициями. Но у таких схем свои пределы скорости и восстановления. В нашем каталоге нужен последний известный статус, поэтому монотонная revision и идемпотентная запись дают понятный минимальный договор.

Ожидаемое состояние после обрыва:

Старый connection/channel завершён
Неизвестный исход ack может дать повтор
Новый consumer использует новый delivery_tag
Inbox узнаёт прежний eventId
Projection не принимает меньшую revision вместо большей

Схема не утверждает, что callback будет вызван ровно один раз после reconnect или в фиксированный момент. Обнаружение обрыва, восстановление broker и доступность хранилища влияют на наблюдение. Инструкция описывает нужные свойства результата, а не подменяет их выдуманным сетевым trace.

Репликация — другая часть устойчивости

Два процесса Python не создают две копии содержимого очереди. В итоговом проекте durable classic queue хранится на одном учебном узле. При потере его диска клиентский reconnect не восстановит исчезнувшее сообщение. Для рабочего требования пережить потерю узла нужно изучать реплицируемые очереди и состояние большинства реплик.

Quorum queues используют другую модель хранения и отказов. Их выбор затрагивает ресурсы, topology и настройки dead lettering, а не только слово в одном аргументе queue_declare. В частности, миграция существующей classic queue должна сохранить важную работу отдельным протоколом. Просто удаление и повторное создание теряет сообщения и не является безопасным обновлением.

Также различайте восстановление consumer и восстановление из исходного журнала. Broker-очередь после ack не хранит безграничную историю для нового получателя. Если надо заново построить каталог, владелец данных должен предоставить снимок или механизм replay известных фактов. В нашем source outbox сохраняется, но автоматического восстановления новых подписчиков из всей истории нет.

Для упражнения поставьте revision 1 в отложенный путь, а revision 2 примените сразу. Объясните результат следующего callback без ссылки на случайный порядок сети. Затем сопоставьте reconnect клиента с потерей носителя broker. Далее введём transactional outbox, чтобы принятое изменение API сохраняло намерение публикации даже до первого сетевого обращения.

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