Продолжаем делиться конспектами Алексея Барабанова, IT-директора «Хлебницы». На этот раз обсудим специфику работы RabbitMQ с высокими нагрузками (High Load) и обеспечением высокой доступности (High Availability). Рассмотрим различные методы увеличения производительности и горизонтального масштабирования, разберём и настроим внутренние инструменты. Также по мере погружения постараемся изучить основные подводные камни всех подходов.

Другие конспекты:

RabbitMQ: терминология и базовые сущности

Как запускать RabbitMQ в Docker

Типовое использование RabbitMQ

RabbitMQ: дополнительные возможности

Shovel

Представляет собой просто интегрированный в RabbitMQ перекладыватель сообщений из одного места в другое (consumer+publisher).

Берёт сообщения из одного RabbitMQ (очереди или exchange) и перекладывает в другой RabbitMQ (exchange или очередь). Даже не обязательно в другой — можно перекладывать и в тот же самый. Работает полностью по протоколу AMQP

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

Также можно запустить shovel на третьей стороне. Например, запустить RabbitMQ вообще без очередей и обработки сообщений — только с shovel. Относительная легковесность RabbitMQ вполне себе это позволяет. Помогает отделить мух от котлет.

Шовелов (shovel) в одном инстансе RabbitMQ может быть несколько.

Динамический shovel настраивается после запуска rabbit через консоль/веб интерфейс/restapi. Статический — через конфиги. У каждого подхода свои плюсы и минусы, зависит от конкретного кейса использования. Динамические шовелы можно выгрузить через export_definitions. Нотация динамических и статических шовелов отличается радикально, хотя набор возможностей схож (статический умеет дополнительно в декларирование и несколько источников).

Если не включать опцию «Add forwarding headers» — работает весьма производительно. Для ускорения рекомендую запускать несколько инстансов одного и того же shovel даже в рамках одного rabbit — 3-4шт вполне себе увеличивают производительность, дальше уже прирост незначительный (упираются в rabbitmq). Похоже, что работают однопоточно.

RabbitMQ весьма скуп на логирование shovel. Часто довольно сложно понять почему он не работает, хотя настроенный shovel какого-то дополнительного обслуживания не требует.

Не умеет ни в какую, даже самую базовую, логику. Имейте ввиду: если вы захотите какое-то шардирование или базовую фильтрацию/дедупликацию, shovel этого не умеет. Надо будет писать полностью своё решение (это не сложно).

Shovel работает в кластере. Получаем все возможности кластеризации сразу, отказ одного сервера кластера не вызывает аварии. Shovel перезапускается на другом узле кластера.

«RabbitMQ для админов и разработчиков»

Federation

По сути у этого плагина есть два отдельных понятия — federation exchange и federation queue.

Сразу хочу сказать, что federation queue я не смог настроить. Вся документация в интернете только теоретическая, нет ни одного скриншота работающего federation queue. При настройке upstream на downstream серверах очередь с аналогичными настройками появляется на upstream, но не происходит никакого консьюминга, никакой балансировки нагрузки. Ничего. Если вдруг у вас есть информация и примеры, как его правильно настроить, буду очень признателен данной информации.

federation exchange — штука уже более реальная, работающая. По сути подключает внутренний exchange downstream сервера к exchange upstream.

Каждый downstream получит копию сообщения пока подключен к upstream. Судя по всему никакого накопления данных для downstream не происходит — пока он не подключен, сообщения он теряет. Настройка Expires и Message TTL не влияет на это поведение.

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

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

High Load

Для начала надо понять что High Load бывает разный. По сути highload — момент, когда ваш сервер перестал справляться с нагрузкой. Это может случится даже на потоке в одно сообщение в секунду. Например, если вы не успеваете его обработать за секунду. И вам нужны какие-то механизмы расшивания производительности — как вертикальные (наращивание мощностей: больше оперативки, больше процессоров), так и горизонтальные (наращивание инстансами).

С какими типами высокой нагрузки мы можем столкнуться:

Большое количество соединений. Когда publisher больше 1000 штук, начинаются проблемы. Надо балансировать.
Большой поток сообщений. Надо потоки делить по rabbit — или балансировкой соединений, или выделением балансирующего rabbit.
Большое количество открытий/закрытий коннектов/каналов. Желательно не допускать такого, но если выбора нет — ставить amqproxy между publisher и rabbit.
Сообщения не успевают обрабатываться. Задача вне рамок этой статьи. Если не получается просто отскалить количество consumer, надо оптимизировать скорость обработки, rabbit тут уже при чем.

Далее обо всём по порядку.

Большое количество соединений от publisher

Когда publisher больше 1000 штук, начинаются проблемы. Можно поднять количество доступных соединений, но, как показывает практика, от такого решения вылазят новые проблемы совсем в неожиданных местах. Правильнее будет поставить балансировщик соединений и запустить несколько инстансов rabbit для приёма сообщений:

В качестве балансировщика соединений я рекомендую использовать haproxy. Я экспериментировал с разными балансировщиками, и только haproxy дал стабильный результат. На практике nginx очень плохо справился с поддержанием соединений AMQP, хотя возможно есть какая-то секретная опция, которая решает эти проблемы.

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

Теперь сообщения со всех publishers будут собраны на разных инстансах RabbitMQ, остаётся их соединить в один общий RabbitMQ.

Для этого у RabbitMQ есть механизмы shovel и federate, оба вам помогут в данной ситуации, но я бы рекомендовал на данном кейсе использовать именно shovel, запущенный на каждом инстансе внешних RabbitMQ (ext 123):

Большой поток сообщений

Если же у вас проблема в производительности RabbitMQ при большом потоке сообщений — применяем аналогичную схему, только не соединяющую все сообщения в один финальный RabbitMQ, а обрабатывающую сообщения на каждом инстансе:

Как вариант, можно поставить один балансирующий RabbitMQ, работающий на самых простых и быстрых механизмах (fanout/direct exchange), и shovel нагребать сообщения из этого rabbit уже в отдельные RabbitMQ с более сложной и менее производительной логикой. Если ограничить длину сообщений со стороны этих внутренних RabbitMQ, можно гибко управлять нагрузкой. Тут shovel логичнее размещать со стороны внутренних RabbitMQ. Так проще будет добавлять в схему новые инстансы RabbitMQ.

Понятное дело, что тут мы упираемся в производительность первого RabbitMQ, но если вам не хватает скоростей до 30000мпс, возможно, вам вообще на таком проекте не подходит RabbitMQ. Ну, или можно разбалансировать LB:

Всё что я говорил выше, работает и для кластеров — все правила аналогичные.

Большое количество открытий/закрытий коннектов/каналов

Желательно, конечно, не допускать такого, но если выбора нет — можно ставить amqproxy между publisher и RabbitMQ. Он примет на себя все пересоздания каналов. В результате RabbitMQ увидит только один канал, в рамках которого происходит публикация.

Важно: consumer не стоит подключать через этот сервис.

Конечно, все эти решения можно и нужно комбинировать в случае необходимости:

AMQProxy умеет работать только с одним энтрипоинтом RabbitMQ, поэтому ставить его надо перед каждым.

Итого: для принятия высокой нагрузки наши основные инструменты — это haproxy как балансировщик нагрузки по шардам, amqproxy (если есть множественное пересоздание соединений для паблиша), а также shovel (или его самописный аналог) для переноса сообщений между инстансами и(или) кластерами RabbitMQ.

High Availability

Отказоустойчивость в RabbitMQ обеспечивается в первую очередь механизмами кластеризации. Реально запуск одного инстанса rabbit — это уже запуск кластера, состоящего из одной ноды. Кластеризация у RabbitMQ, что называется «by design».

Классическая кластирезация в rabbit — это репликация очередей.

Для каждой очереди выбирается мастер нода, которая ответственна за обработку именно этой очереди. Если не настроить явным образом, очередь будет располагаться и работать только на этой ноде. Для классической репликации необходимо для очередей указывать дополнительные policy. В результате мы получим работу очереди на мастер ноде и репликацию состояния этой очереди на mirror-ноды(реплики). Количество реплик задаётся через policy для каждой очереди индивидуально.

Со стороны приложения нам не нужно узнавать какая из нод мастер. Работать можно с любой — запросы будут автоматически спроксированы на мастер-ноду.

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

Работа кластера требует связности между нодами не более 30мс, что делает невозможным работу кластера между удалёнными ДЦ — для дублирования сообщений используются уже известные нам механизмы shovel и federation.

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

Сетевая схема

Упрощенно кластер из трёх нод можно изобразить так:

С publisher:

В случае выхода из строя 1й ноды работа publisher становится невозможной (без доработки коннектора):

Поэтому ставим балансировщик (consumer подключаем аналогично через балансировщик):

Policy-HA

Давайте пройдёмся по аргументам очередей, влияющиx на их поведение в кластере. Во первых ha-mode указывает политику назначения реплик по нодам.

exactly — просто по количеству реплик (мастер тоже считается репликой, поэтому 1 — это ноль реплик, 2 — одна реплика и тд). Наиболее рекомендуемое значение (количество задается через аргумент ha-params);
all — просто на все ноды кластера;
nodes — перечисление нод, на которых требуется запустить реплики очередей (имена задаются через аргумент ha-params);

ha-sync-mode отвечает за добавление реплик:

manual — ручное управление, зеркала добавляются, только когда основная очередь пуста или явной инициализацией процесса синхронизации. Значение по умолчанию.
automatic — полностью автоматическая синхронизация. Есть кейсы, когда она может выстрелить в ногу, но в 90% случаев лучше использовать именно её.

Логическая схема

Вот наш кластер, но уже применительно к очередям. Мастер очереди по факту создаются там, куда на момент декларирования был подключен канал (зависит от настроек балансировки). Если вы хотите явно создать мастер на конкретной ноде, можно подключиться к веб-интерфейсу этой ноды и создать очередь оттуда. Вот, например, мы создали 4 очереди — 1 и 4 на первой ноде, 2 на второй, 3ю очередь — на третьей ноде. Пока мы не применим policy, никакой репликации и отказоустойчивости у нас не появится.

Добавляем policy:

ha-mode: exactly

ha-sync-mode: automatic

ha-params сделаем разными для разных очередей:

для первой и второй очереди — ha-params: 3 — как мы видим появилось две реплики (итого три ноды заняты обслуживанием этой очереди);
для 3й очереди ha-params: 2 — значит добавляется только одна реплика;
для 4й ha-params: 1, значит, что по факту никакой репликации не будет.

А дальше давайте всё ломать.

Что же произойдет если, к примеру, 1-ая нода выйдет из строя?

Для первой очереди инициируется выбор мастер-ноды. Например, пусть ей станет 2я нода, а вот 4 очереди логично не повезло — она перестанет быть доступной:

Следом, чтобы жизнь не казалась сказкой, падает 3 нода:

Для третьей очереди мастером переизбирается также 2 нода. Все три очереди работают на второй.

Допустим, что нам повезло, и 1 нода ожила сама (ну или мы ей помогли):

1, 2 и 3 очереди заново среплицируются с мастера и станут репликами, 4-ая очередь оживет и останется мастером на этой ноде:

После синхронизации первые две ноды перейдут в стандартный режим работы. А теперь нам удалось запустить 3 ноду, но стейт на ней (например) был полностью утрачен:

Запустится репликация 1 и 2 очередей (тк для третьей необходимое кол-во реплик уже достигнуто):

И вот результат двух отключений нод. Теперь 1, 2 и 3 очереди работают на 2 ноде — это не изменится само, только если вы явным образом не переключите мастер на другую ноду, или же 2 нода не прикажет долго жить (иногда для ребалансировки проще всего бывает именно перезагрузить самую живучую ноду).

«RabbitMQ для админов и разработчиков»

Разбираемся с RabbitMQ: High Availability и High Load