Привет, Хабр! Я Женя, архитектор интеграционной платформы в Точке, отвечаю за асинхронный обмен сообщениями между внутренними сервисами, за ESB и за брокеры сообщений.

В прошлом году я выступил на внутреннем митапе с докладом, в котором постарался кратко, последовательно и местами рок-н-ролльно изложить основные моменты, о которых полезно помнить при использовании RabbitMQ, если важны стабильность обмена и сохранность данных. Эта статья — адаптация доклада.

В первую очередь материал рассчитан на разработчиков, которым ещё не приходилось погружаться в тонкости работы с RabbitMQ или использовать его вообще. Более опытным читателям статья может пригодиться в качестве компактной и упорядоченной выжимки из уже знакомых англоязычных статей, вебинаров и многочисленных страниц документации.

Речь пойдёт о надёжной работе клиентских приложений с брокером, и под ней я подразумеваю отказоустойчивый обмен сообщениями с гарантией доставки «как минимум однократно» (at-least-once).

Мы разберём:

1. Что делать, чтобы сообщения летали без перебоев и не терялись. (Спойлер: делать нужно много чего, потому что кролик с настройками по умолчанию очень любит выбрасывать сообщения и занимается этим при любой возможности, причём молча =)

2. Как при этом оптимизировать производительность и потребление ресурсов.

Чтобы выстроить повествование, представим, как среднестатистическое приложение работает с кроликом: сначала оно подключается к брокеру и объявляет топологию маршрутизации, потом публикует сообщения и потребляет их, а за всем этим (хочется верить) кто-нибудь присматривает. Давайте пройдём все эти шаги по порядку и рассмотрим каждый с точки зрения надёжности.

1. Подключение

1.1. Протокол

RabbitMQ поддерживает несколько протоколов обмена сообщениями. Чтобы не перегружать статью, мы рассмотрим два из них: «встроенный» бинарный AMQP 0-9-1 и простой текстовый STOMP, подключаемый через плагин.

AMQP 0-9-1 оптимален в плане производительности и предоставляет максимальный доступ к возможностям брокера, кролик создавался как его реализация. Ниже по тексту под AMQP я буду подразумевать именно AMQP 0-9-1. STOMP в свою очередь предельно прост и лёгок в имплементации — при желании можно пообщаться с брокером через telnet. В плане надёжности протоколы почти одинаковы. Чуть ниже расскажу, почему «почти».

1.2. Установка соединения

Установка AMQP-соединения с брокером требует 7 TCP-пакетов (при использовании TLS все 12), а каждое установленное соединение использует минимум 100 Кб оперативной памяти, поэтому держать много соединений или постоянно открывать-закрывать их накладно. Чтобы решить эту проблему, протокол AMQP вводит понятие каналов (channels). Это лёгкие логические соединения «внутри» соединений, именно в них идёт вся работа.

Важно помнить, что и соединения, и каналы рассчитаны на то, чтобы «жить» долго и использоваться повторно. В первую очередь, конечно, соединения, но каналы по возможности тоже. Рассмотрим пример: если на публикацию каждого сообщения открывать новое соединение и канал, а потом закрывать их, то каждая такая публикация будет использовать 19 TCP-пакетов вместо 1. Клиент и брокер будут каждый раз заново договариваться о параметрах соединения, выделять ресурсы, а потом высвобождать их. В результате пострадает производительность.

По возможности стоит придерживаться правила «один процесс — одно соединение». При этом потоки исполнения внутри процесса открывают собственные каналы и используют их на протяжении всего жизненного цикла. Использовать один канал несколькими потоками не рекомендуется — в большинстве клиентов каналы не потокобезопасны. Иногда нужны два соединения на процесс, этот случай мы разберём ниже, в разделе «Публикация сообщений». Если же соединений больше, получаем «лишнее» потребление ресурсов и на клиенте, и на брокере.

1.3. Проверка работоспособности соединения

Если с соединением что-то пошло не так, для обеспечения стабильности обмена важно узнать об этом как можно раньше. Для этого используется механизм сердцебиений (heartbeats). При установке соединения клиент и брокер договариваются о некотором значении таймаута (timeout), а потом обмениваются фреймами с интервалом, равным примерно половине таймаута. После двух пропущенных фреймов соединение закрывается.

Если задать слишком низкий таймаут, будут ложные срабатывания. Если задать слишком высокий, на обнаружение проблем уйдёт больше времени. Таймауты в пределах 30–60 секунд выглядят подходящими для большинства случаев.

Очень важно не путать таймаут и интервал. По названию методов и параметров в клиентских библиотеках не всегда ясно, что именно задаётся, но чаще задаётся таймаут. В этом случае, если задать значение 30 секунд, а потом слать фреймы раз в 30 секунд, то соединения будут постоянно закрываться. Если задан таймаут 30 секунд, фреймы нужно слать раз в 15 секунд.

Если не использовать механизм сердцебиений совсем, то есть шанс накопить пачку «мёртвых» соединений и получить отказ при очередной попытке подключения, если на брокере настроен лимит на количество соединений.

1.4. Восстановление соединения

Способность приложения автоматически переподключаться к брокеру при закрытии или обрыве соединения позволяет восстановить работу без ручного вмешательства и тем самым сократить простой в обмене. Это умение особенно важно, если кролик развёрнут в облаке, где отдельные узлы могут уходить на обслуживание в любое время дня и ночи.

В случае RabbitMQ переподключение предполагает восстановление соединения и каналов, а также связанных с ними обработчиков и настроек.

Некоторые клиентские библиотеки инкапсулируют логику автоматического переподключения. В этом случае достаточно убедиться, что опция включена, а конкретные параметры (например, задержка между попытками) соответствуют потребностям приложения. Остальные библиотеки, как правило, предоставляют методы для выполнения отдельных шагов переподключения и рекомендации по их использованию; в этом случае механизм переподключения придётся реализовать самостоятельно на уровне приложения.

2. Топология маршрутизации

Топология маршрутизации в терминах RabbitMQ — это совокупность обменников (exchanges), очередей (queues) и связей (bindings) между ними. Она определяет правила маршрутизации сообщений на брокере: сообщения приходят в обменники, а оттуда направляются в связанные с ними очереди (или другие обменники). Топология зачастую объявляется клиентом, и то, насколько она будет надёжна, в таких случаях почти полностью зависит от клиента.

При объявлении топологии действуют следующие правила:

• если элементов на брокере нет, они создаются;

• если элементы есть, а их настройки совпадают с заявленными, ничего не происходит;

• если элементы есть, но настройки отличаются, возникает ошибка.

2.1. Общие рекомендации

Начнём с простого, но без преувеличения жизненно важного момента: при перезапуске брокера «выживут» только те элементы топологии, которые были объявлены как durable. Например, если классическая очередь не была durable, то как бы надёжно мы ни публиковали в неё сообщения, при перезапуске брокера она помашет нам ручкой и исчезнет вместе со всеми сообщениями. Only durable left alive =)

Объявлять топологию надёжней в коде приложения, не вручную — так мы исключаем человеческий фактор и можем в случае чего быстро и точно её восстановить.

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

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

2.2. Защита от потери сообщений

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

Первый способ — alternate exchange. Это обычный обменник, который можно указать у исходного обменника в свойстве alternate-exchange, чтобы сообщения, которые не удалось маршрутизировать, направлялись в него, а не в /dev/null. Конечно же, к альтернативному обменнику должна быть привязана очередь для складирования этих сообщений, откуда их можно будет переложить в другую очередь или вычитать обработчиком по умолчанию. Надежнее всего использовать в качестве альтернативного обменник с типом fanout — он направит в привязанную очередь любые сообщения вне зависимости от значений их заголовков и ключей маршрутизации.

Второй способ — это флаг mandatory=true, который можно задать при публикации сообщения, чтобы брокер вернул его обратно, если не найдёт подходящую очередь. Тонкости работы этого флага описаны в разделе «Публикация сообщений».

2.3. Dead lettering

Сообщения иногда «умирают» — при отказе в обработке на потребителе, по истечении срока жизни или из-за переполнения очереди при определённых настройках. Разумеется, по умолчанию кролик выбрасывает такие сообщения, но мы можем попросить его так не делать. Поможет нам в этом dead letter exchange, для друзей просто DLX.

Очередь можно настроить на отправку «умерших» сообщений в заданный обменник. Для этого достаточно указать имя обменника в её свойстве dead-letter-exchange. Дополнительно можно переопределить ключ маршрутизации с помощью свойства dead-letter-routing-key. Конечно же, обменник с указанным именем должен существовать и иметь подходящую связь хотя бы с одной очередью, иначе кролик продолжит выбрасывать сообщения (частности про at-least-once dead lettering оставим за рамками статьи).

Обменники, играющие роль DLX, технически ничем не отличаются от других обменников — они создаются и работают стандартно. В качестве DLX можно использовать даже обменник по умолчанию (default exchange). Для этого свойство dead-letter-exchange исходной очереди нужно задать пустым, а в dead-letter-routing-key указать имя целевой очереди — той, в которую должны идти сообщения.

Отлично, мы настроили DLX и больше не теряем «умершие» сообщения, теперь они складируются в отдельные очереди, но! Если с такими сообщениями ничего не делать, ситуация будет мало чем отличаться от потери. Поэтому важно замечать их появление (например, с помощью систем мониторинга и алертинга) и реагировать — разбираться, почему сообщения выпали в DLX, нужны ли доработки на потребителе, нужна ли повторная обработка.

Для повторной обработки достаточно вернуть сообщения в исходную очередь. Кстати, кролик позволяет делать это автоматически, с помощью того же механизма DLX. Такой подход может быть удобен, когда ошибки обработки носят временный характер, и для успеха достаточно просто повторить попытку. У очереди, куда приходят «умершие» сообщения, можно указать время их жизни (message-ttl), а уже упомянутые свойства dead-letter-exchange и dead-letter-routing-key задать так, чтобы сообщения возвращались в исходную очередь. В результате сообщения будут автоматически отправляться на повторную обработку с заданной задержкой.

3. Публикация сообщений

Итак, мы подключились к брокеру, объявили надёжную топологию и теперь хотим опубликовать сообщение. Давайте рассмотрим, как защититься от потери сообщений на этом этапе.

3.1. Запись на диск

Во-первых, нужно попросить кролика записать сообщение на диск. Классические очереди по умолчанию хранят сообщения только в оперативной памяти и теряют их при перезапуске брокера. Персистентность сообщений в AMQP задаётся через указание delivery-mode=2 в свойствах basic, а в STOMP — через использование заголовка persistent: true во фреймах SEND.

3.2. Подтверждение приёма

Во-вторых, нужно попросить кролика подтвердить приём сообщения и дождаться этого подтверждения. По умолчанию подтверждений нет, поэтому при проблемах с брокером или соединением сообщение просто потеряется. В AMQP нужно использовать publisher confirms и ждать подтверждения от брокера. В STOMP нужно задавать заголовок receipt во фрейме SEND и ждать от брокера фрейм RECEIPT.

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

Длительность ожидания перед повтором задаётся исходя из окружения и требований к обмену. Чем меньше длительность ожидания, тем быстрее восполняются потенциальные потери, но и тем вероятнее дублирование сообщений из-за проблем с производительностью у одной из сторон или со связью между ними.

3.3. Защита от отсутствия маршрута

А теперь вопрос на засыпку =) Как думаете, что произойдёт, если мы попросим кролика записать сообщение на диск и отправить нам подтверждение, но кролик не найдёт подходящий маршрут? Бинго, он выбросит сообщение. Более того, он ещё и подтверждение пришлёт, что всё хорошо. По сути подтверждение говорит о том, что кролик поделал с сообщением всё, что хотел. И в данном конкретном случае он хотел его выбросить.

Защититься от отсутствия маршрута на уровне публикации позволяет специальный флаг, который есть только в AMQP (именно поэтому рассматриваемые протоколы «почти» одинаковы в плане надёжности). С помощью флага mandatory=true во фрейме basic.publish можно попросить кролика возвращать сообщение, если оно не попало ни в одну очередь. Важный момент: брокер не отказывает, он по-прежнему подтверждает приём, но в дополнение к этому возвращает сообщение в отдельном фрейме basic.return. Если на эти фреймы не реагировать, смысл флага теряется, поэтому при его использовании клиент должен уметь обрабатывать возвраты — например, публиковать сообщение в другой обменник. Я бы рассматривал использование флага mandatory как опциональную меру для исключительной надёжности.

3.4. Отдельное соединение для публикации

А теперь вернёмся к правилу «один процесс — одно соединение» и рассмотрим его расширение, которое связано с публикацией сообщений.

У кролика есть защитный механизм flow control, искусственно «притормаживающий» соединения со слишком активной публикацией. Если эти же соединения используются потребителями, у них возникают трудности с отправкой подтверждений и, как следствие, проблемы с производительностью. В худшем случае гремучая смесь интенсивной публикации и замедленного потребления может привести к переполнению очередей и истощению ресурсов на брокере.

Отсюда возникает рекомендация использовать отдельные соединения для публикации и потребления, а значит для процессов, которые занимаются и тем, и другим, правило расширяется до «один процесс — два соединения» (одно для публикации, другое для потребления).

4. Потребление сообщений

При потреблении сообщений в контексте надёжности важны ответственность и умеренность, если вы понимаете, о чём я =) Давайте внимательно рассмотрим оба аспекта.

4.1. Подтверждение обработки

По умолчанию кролик удаляет сообщения, как только поместил их в TCP-буфер на отправку потребителю, поэтому проблемы с машиной брокера, с соединением или с потребителем легко могут привести к потере данных. Чтобы этого избежать, нужно попросить брокер удалять сообщения только после получения подтверждения от потребителя. В AMQP режим подтверждений включается свойством no-ack=false в basic.consume. В STOMP — заголовком ack: client[-individual] во фрейме SUBSCRIBE.

Но мало просто включить режим, важно ещё отправлять подтверждения (acks) в правильный момент — не сразу при получении, а только после того, как успешно прошла вся обработка. Например, после сохранения в базу или пересылки в другую очередь. В общем, только тогда, когда потребитель полностью принял ответственность за сообщение на себя.

Когда режим подтверждений включён, брокер ждёт от потребителя либо ack (в случае успеха), либо nack/reject (при отказе). Длительность ожидания ограничена на брокере параметром consumer_timeout (30 минут по умолчанию), затем канал потребителя закрывается с ошибкой, а сообщение возвращается в очередь. Во время ожидания сообщение находится в статусе unacked, который подводит нас к теме умеренности.

4.2. Умеренное потребление

По умолчанию кролик пытается выдать потребителю как можно больше сообщений. Потребителю все эти сообщения нужно держать в буфере, поэтому от такой щедрости он может переесть ресурсов и убиться об OOM. Чтобы не допустить трагедии, на потребителе есть настройки, которые ограничивают выдачу: prefetch-count (по количеству сообщений) и prefetch-size (по объёму данных). Когда лимит достигнут, брокер прекращает выдачу до получения подтверждений от потребителя. Если режим подтверждений не включён, ограничение просто не действует.

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

Главный антипаттерн — это не задавать префетч вообще (ООМ Killer не дремлет), и по умолчанию он, конечно же, не задан. Задавать префетч лучше на уровне потребителя, то есть с флагом global: false. Этот вариант совместим со всеми типами очередей RabbitMQ и работает быстрее, чем префетч на уровне канала. В AMQP ограничение по количеству сообщений задаётся с помощью свойств prefetch-count=N и global=false в basic.qos. В STOMP — c помощью заголовка prefetch-count: N во фрейме SUBSCRIBE.

Как выбрать prefetch count?

Осталось выбрать значение для N. А как его выбрать? Душа просит формулу. И формула есть. Вот она:

В числителе стоит длительность полного цикла (доставка сообщения от брокера потребителю, обработка потребителем, доставка подтверждения от потребителя брокеру), а в знаменателе — длительность обработки потребителем.

Формула не годится для выбора окончательного значения, потому что предполагает постоянное время обработки сообщений и неизменные условия на сети, но помогает найти отправную точку. Она показывает, какой префетч должен задать идеальный потребитель в идеальной сети, чтобы к тому моменту, когда он обработает последнее сообщение полученной пачки, от брокера пришло следующее сообщение. В этом случае и потребитель не простаивает в ожидании, и сообщения у него в буфере не копятся.

В реальных условиях всё немного сложнее: скорость обработки варьируется, состояние сети меняется, к обмену предъявляются разные требования. Поэтому префетч обычно подбирается под конкретную ситуацию.

При подборе можно ориентироваться на несколько факторов:

• если префетч не задан или задан слишком высокий, то потребитель может упереться в вычислительные ресурсы;

• если префетч задан низкий, а обработка быстрая, то потребитель будет простаивать в ожидании новых сообщений от брокера;

• если префетч задан высокий, а обработка медленная, то сообщения будут лежать в буфере потребителя, занимать там место и ждать обработки, хотя могли быть доставлены другим, менее занятым потребителям.

Подробности с наглядными картинками можно найти в статье на CloudAMQP.

5. Мониторинг и алертинг

И последний по счёту (но не по значимости) пункт. Я просто оставлю здесь минимальный набор метрик, за которыми, на мой взгляд, полезно следить в контексте сохранности данных и стабильности обмена.

1. Объём данных в очереди
   Очередь может быть ограничена по объёму содержащихся в ней данных, а переполнение при определённых настройках может приводить к удалению сообщений. Метрика предупредит о приближении к заданному лимиту.

2. Количество сообщений в очереди
   Скопление сообщений может говорить о проблемах на потребителях. Например, потребители вышли из строя, деградировали или их текущей пропускной способности перестало хватать для возросшей нагрузки.

3. Количество потребителей на очереди
   Метрика поможет заметить выход обработчиков из строя, даже если в обменах тишина и очереди не растут — то есть до того, как это станет проблемой.

4. Наличие немаршрутизируемых сообщений
   Показатель позволит узнать о потере сообщений из-за недочётов в топологии.

5. Наличие сообщений в DLX
   Метрика поможет заметить проблемы в обработке сообщений и среагировать на них.

Этот минимальный набор можно использовать в качестве отправной точки и расширять под конкретные потребности.

Официальная документация рекомендует использовать для мониторинга RabbitMQ популярный тандем Prometheus и Grafana. В подробном руководстве среди прочего можно найти внушительный список метрик, предоставляемых плагином rabbitmq_prometheus, и ссылку на список официальных дашбордов для Grafana.

Итого

Вот мы и разобрали основные моменты, которые полезно иметь в виду при работе с RabbitMQ, если важны стабильность обмена и сохранность данных. Надёжная доставка сообщений находится в общей ответственности брокера и клиентских приложений, которые их публикуют и потребляют. RabbitMQ предоставляет целый арсенал механизмов для защиты от потери сообщений и соблюдения гарантий доставки, однако почти все они по умолчанию выключены в угоду простоте и производительности и требуют не только явного включения, но и правильной настройки.

В заключение отмечу, что надёжность, конечно же, не бесплатна. Во-первых, как вы уже убедились, нужно приложить немало усилий, чтобы отвадить кролика выбрасывать сообщения и учесть остальные тонкости. Во-вторых, запись сообщений на диск и отправка дополнительных фреймов по сети неизбежно влияют на производительность. В-третьих, при гарантированной доставке возникает вероятность дублирования сообщений, которую нужно учитывать при обработке на потребителях, но это уже совсем другая история =)