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

Меня зовут Александр Борецкий, я архитектор в Т1 Облако. Поделюсь своим опытом работы с самыми популярными из брокеров сообщений — Kafka и RabbitMQ. Расскажу, как выбрать и настроить подходящий брокер, а также какие архитектурные особенности есть у каждого из них.

Напомню, что брокеры сообщений:

• обеспечивают надёжную доставку данных и интеграцию различных частей систем друг с другом;

• определяют с помощью механизма маршрутизации, куда и каким образом нужно доставить сообщения;

• размещают сообщения в очередях, где те хранятся до обработки получателем;

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

• распределяют нагрузку между множеством сервисов и поддерживают высокую производительность всей системы.

Популярные брокеры могут «переваривать» до нескольких миллионов сообщений за раз и гарантировать их доставку. Многие большие компании пользуются сразу несколькими брокерами сообщений. Как правило, это и Kafka, и RabbitMQ. Рассмотрим каждый поподробнее.

Kafka

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

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

Отличительные особенности Kafka: параллельная обработка данных, высокая пропускная способность, масштабируемость.

Установка Kafka

Для начала посмотрим, как развернуть минимальную конфигурацию ноды Kafka на примере версии брокера 3.8.0 и хоста Ubuntu 22.04.

1. Проверим наличие Java на сервере или установим его (Kafka требует Java 8+):

   java -version # проверка версии
   sudo apt update && sudo apt install -y openjdk-11-jre #установка

2. Скачаем и установим Kafka в папку /opt/kafka:

   wget https://downloads.apache.org/kafka/3.8.0/kafka_2.13-3.8.0.tgz #скачиваем дистрибутив
   tar -xzf kafka_2.13-3.8.0.tgz #распаковываем
   mv ./kafka_2.13-3.8.0 /opt/kafka/ #копируем в каталог
   cd /opt/kafka #переходим в него

3. Настроим конфигурацию сервера, используя nano config/server.properties, раскомментируем и дополним следующие параметры:

   listeners=PLAINTEXT://:9092
   advertised.listeners=PLAINTEXT://#ip cервера#:9092
   log.dirs=/opt/kafka/logs

4. Создадим сервис kafka.service, используя sudo nano /etc/systemd/system/kafka.service. Вставляем в него:

   [Unit]
   Description=Apache Kafka Server
   After=network.target zookeeper.service
   
   [Service]
   User=root
   ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
   ExecStop=/opt/kafka/bin/kafka-server-stop.sh
   Restart=always
   
   [Install]
   WantedBy=multi-user.target

5. Даже для одной ноды Kafka необходим координатор ZooKeeper или KRaft для хранения метаданных. Клиент Kafka поставляется со встроенным ZooKeeper (для тестов и разработки), и мы будем использовать его в этой инсталляции.

   Создадим сервис zookeeper.service, используя sudo nano /etc/systemd/system/zookeeper.service. Вставляем в него:

   [Unit]
   Description=Apache ZooKeeper
   After=network.target
   
   [Service]
   Type=simple
   User=kafka
   Group=kafka
   ExecStart=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties
   ExecStop=/opt/kafka/bin/zookeeper-server-stop.sh
   Restart=on-failure
   RestartSec=5
   
   [Install]
   WantedBy=multi-user.target

6. И запустим сервисы ZooKeeper и Kafka:

   sudo systemctl daemon-reload
   sudo systemctl enable zookeeper
   sudo systemctl start zookeeper
   sudo systemctl status zookeeper
   sudo systemctl enable kafka
   sudo systemctl start kafka
   sudo systemctl status kafka

Видим, что наши сервисы успешно запущены:

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

Далее все примеры команд будут основываться на этом сервере и с текущим расположением каталогов.

Для повышенной отказоустойчивости и производительности несколько брокеров можно объединить в кластер; для координации между брокерами используется ZooKeeper. Начиная с версии 3.0, Kafka также поддерживает использование собственного координатора — KRaft (Kafka Raft Metadata).

Для отказоустойчивой координации брокеров рекомендуется использовать не менее трёх компонентов ZooKeeper и располагать их отдельно от брокеров.

Пример базовой конфигурации ZooKeeper:

tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181
initLimit=5
syncLimit=2
server.1=zookeeper1:2888:3888
server.2=zookeeper2:2888:3888
server.3=zookeeper3:2888:3888

В конфигурационном файле каждого брокера Kafka также необходимо прописать подключение ко всем хостам ZooKeeper:

zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181 #Список узлов ZooKeeper 
zookeeper.session.timeout.ms=6000 #Время жизни сессии ZooKeeper.
zookeeper.connection.timeout.ms=10000 # Таймаут подключения к ZooKeeper.

Особенности

Теперь детально рассмотрим особенности брокера Kafka.

Топик

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

Пример создания топика через CLI (kafka-topics.sh):

# Создание топика с тремя партициями 
bin/kafka-topics.sh --create --topic my-topic --partitions 3  --bootstrap-server localhost:9092

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

bin/kafka-configs.sh --alter --entity-type topics --entity-name my-topic --add-config retention.ms=604800000 --bootstrap-server localhost:9092

Или вовсе удалить при необходимости:

bin/kafka-topics.sh --delete --topic my-topic --bootstrap-server localhost:9092

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

Партиция

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

bin/kafka-topics.sh --alter --topic my-topic --partitions 5 --bootstrap-server localhost:9092

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

Оффсет

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

Автоматическое сохранение оффсетов задано по умолчанию в свойствах топика:

enable.auto.commit=true  
auto.commit.interval.ms=5000

В данном случае оффсеты сохраняются каждые 5 секунд.

Если нужно гарантировать сохранение оффсетов именно после обработки конкретного количества сообщений, то лучше использовать ручной режим (enable.auto.commit=false) и сохранять оффсеты вручную.

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

• Kafka Manager — инструмент для управления кластером Kafka, мониторинга топиков, партиций и производительности.

• Confluent Control Center — расширенный инструмент для мониторинга и управления Kafka, включая визуализацию потоков данных, мониторинг задержек и производительности.

• JMX. Kafka предоставляет метрики через JMX, которые можно собирать с помощью таких инструментов, как Prometheus и Grafana.

• Kafka Monitoring Plugins — плагины для сбора метрик из Kafka, например, kafka_exporter для Prometheus.

Преимущества

Теперь, когда мы рассмотрели архитектуру и особенности Kafka, остановимся на её преимуществах.

Производительность

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

При отправке сообщения в брокер для гарантированного порядка их обработки в рамках одной партиции можно указать ключ (key):

producer.send('topic1', key=123, value=message)

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

producer.send('topic1', partition=2, value=message)

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

Балансировка

Тоже важная особенность работы Kafka. Балансировка осуществляется с помощью групп потребителей, когда партиции автоматически распределяются между читателями топиков в рамках одной группы. Её задают параметром partition.assignment.strategy. Есть три варианта распределения:

• RangeAssignor используется по умолчанию и распределяет партиции между потребителями на основе диапазонов. Например, если есть 6 партиций и 3 потребителя, то первый потребитель получит партиции 0-1, второй — 2-3, третий — 4-5.

• RoundRobinAssignor распределяет партиции по кругу между всеми потребителями. Например, если есть 6 партиций и 3 потребителя, то первый потребитель получит партиции 0, 3, второй — 1, 4, третий — 2, 5.

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

Отказоустойчивость

Отказоустойчивость обеспечивается с помощью репликации данных, настройки параметров подтверждения и использования механизма idempotent producer. Репликация задаётся параметром replication.factor при создании топика, например:

kafka‑topics.sh ‑create ‑bootstrap‑server localhost:9092 ‑topic my‑topic ‑partitions 3 ‑replication‑factor 3

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

Подтверждение записи

Это механизм, когда отправитель сообщения может считать запись успешно отправленной. Задаётся параметрами acks и min.insync.replicas. acks настраивается в конфигурации отправителя, а min.insync.replicas — в топике.

acks имеет три варианта подтверждения:

• 0 — отправитель не ждёт подтверждения от брокера. Это самый быстрый, но наименее надёжный вариант.

• 1 — подтверждение отправляется после записи данных в лидера партиции. Если лидер выйдет из строя до синхронизации с репликами, то данные могут быть потеряны.

• All — подтверждение отправляется только после того, как все реплики получили данные. Это самый надёжный вариант, но и самый медленный.

min.insync.replicas определяет минимальное количество реплик, которые должны быть синхронизированы с лидером партиции для подтверждения успешности записи.

Механизм Idempotent producer

Он гарантирует, что каждое сообщение будет записано в Kafka ровно один раз. Задаётся включением параметра [«enable.idempotence», true], также обязательным условием является acks=all.

Пример создания:

producer = KafkaProducer(
bootstrap_servers='localhost:9092', # Адрес сервера Kafka
acks='all', # Подтверждение, что все реплики получат сообщение
enable_idempotence=True # Включение механизма идемпотентности
)

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

Масштабируемость

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

По умолчанию новые брокеры не получают старые партиции. Для распределения нагрузки используется скрипт kafka‑reassign‑partitions.sh в несколько шагов:

1. Для получения текущей схемы партиций используем:

   bin/kafka‑reassign‑partitions.sh ‑zookeeper <zookeeper_host>:2181 ‑generate ‑topics‑to‑move‑json‑file topics.json ‑broker‑list "<список_брокеров>"

2. После чего запускаем пересоздание:

   bin/kafka‑reassign‑partitions.sh ‑zookeeper <zookeeper_host>:2181 ‑execute ‑reassignment‑json‑file reassign.json

3. Если необходимо распределить реплики на новый брокер, то необходимо обновить replication.factor:

   bin/kafka‑topics.sh ‑alter ‑topic <topic_name> ‑partitions <new_partition_count> ‑bootstrap‑server <bootstrap_servers>

После этого новый брокер станет частью Kafka‑кластера.

Постоянное хранение данных

Kafka может долго хранить данные благодаря параметрам удержания. retention определяет, сколько времени данные будут сохраняться в топике.

Пример настройки длительности хранения:

bin/kafka‑configs.sh ‑alter ‑entity‑type topics ‑entity‑name my‑topic ‑add‑config retention.ms=604800000 ‑bootstrap‑server localhost:9092

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

bin/kafka‑topics.sh ‑create \
‑bootstrap‑server <broker‑address> \
‑topic event‑store‑topic \
‑partitions 6 \
‑replication‑factor 3 \
‑config cleanup.policy=compact \ # Включает архивацию, которая сохраняет последние значения для каждого ключа
‑config retention.ms=-1 \ #бесконечное хранение событий. 
‑config segment.bytes=1073741824 \ # Размер сегмента журнала в байтах (1 ГБ). Компактизация работает на уровне сегментов, поэтому большие сегменты могут увеличить задержку архивации.
‑config min.compaction.lag.ms=5000 # Минимальное время (в миллисекундах), которое должно пройти после записи события, прежде чем оно будет архивировано. Это предотвращает слишком частую компактизацию.

Интегрируемость

Это свойство позволяет воспользоваться богатой экосистемой инструментов для расширения функциональности и интеграции с другими системами.

Kafka Streams API — это библиотека для анализа и обработки потоковых данных в реальном времени.

Пример использования:

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");

// Преобразование данных
KStream<String, String> transformed = source.mapValues(value -> value.toUpperCase());

// Отправка результатов в новый топик
transformed.to("output-topic");

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Интеграция с Confluent Platform позволяет использовать дополнительные компоненты, такие как:

• Kafka Connect — для интеграции с внешними системами (например, базами данных, файловыми системами);

• KSQL — для выполнения SQL‑запросов к потоковым данным в реальном времени;

• Schema Registry — для управления форматами сообщений.

Защита данных

Она обеспечивается с помощью шифрования каналов связи и аутентификации пользователей. Для включения шифрования данных между брокерами внутри кластера необходимо выпустить цепочку SSL‑сертификатов и подгрузить их в параметры брокеров:

# Включение SSL для внутреннего общения между брокерами
listeners=SSL://:9093  # Порт для SSL-соединений между брокерами
advertised.listeners=SSL://<broker-host>:9093  # Адрес, который будет объявлен клиентам

# Путь к сертификатам для SSL
ssl.keystore.location=/path/to/kafka.server.keystore.jks  # Путь к keystore-файлу брокера
ssl.keystore.password=<keystore-password>  # Пароль для доступа к keystore
ssl.key.password=<key-password>  # Пароль для ключа в keystore
ssl.truststore.location=/path/to/kafka.server.truststore.jks  # Путь к truststore-файлу брокера
ssl.truststore.password=<truststore-password>  # Пароль для truststore

# Протокол и алгоритм шифрования
security.inter.broker.protocol=SSL  # Протокол для общения между брокерами
ssl.enabled.protocols=TLSv1.2,TLSv1.3  # Поддерживаемые версии протоколов TLS
ssl.cipher.suites=TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384  # Алгоритмы шифрования

# Требовать проверку сертификата клиента (необязательно)
ssl.client.auth=required  # Если нужно требовать сертификат от клиентов

Если необходимо зашифровать также подключения клиентов (продюсеров/потребителей), то в конфигурацию добавляем параметр проверки сертификатов подключения: ssl.client.auth=required.

Со стороны клиента также нужно подгрузить сертификаты и включить SSL‑подключение:

security.protocol=SSL  # Использовать SSL для соединения с брокером
ssl.truststore.location=/path/to/client.truststore.jks  # Путь к truststore клиента
ssl.truststore.password=<truststore-password>  # Пароль для truststore клиента

# Если требуется аутентификация клиента
ssl.keystore.location=/path/to/client.keystore.jks  # Путь к keystore клиента
ssl.keystore.password=<keystore-password>  # Пароль для keystore клиента
ssl.key.password=<key-password>  # Пароль для ключа в keystore клиента

RabbitMQ

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

Как пример, RabbitMQ подойдёт для отправки push‑сообщений, SMS и уведомлений клиентам из систем бронирования билетов, онлайн‑магазинов и других приложений, а также для гарантированного обмена сообщениями между критичными системами и микросервисами в банках, телекоммуникационных и аэрокосмических компаниях.

Чтобы сообщения гарантированно дошли до конкретных получателей, RabbitMQ использует механизм гибкой маршрутизации. Например, мобильное приложение для управления умным домом сначала отправляет сообщения в обменник (Exchange), который маршрутизирует каждое в соответствующую очередь на основе заранее определённых условий. То есть сообщения попадают в очередь для конкретного получателя, в данном случае — для каждого подключённого устройства: светильнику отправляется сообщение с задачей включить свет, кофеварке — начать варить кофе.

Отличительные особенности RabbitMQ: гибкая маршрутизация, устойчивость к сбоям, гарантированная доставка, отложенная обработка сообщений.

Установка RabbitMQ

Не будем подробно останавливаться на установке самого RabbitMQ. Он содержится в основном репозитории Ubuntu 22.04, поэтому его можно легко установить через apt install:

1. Установим RabbitMQ‑server:

   sudo apt install ‑y rabbitmq‑server

2. Проверим, что всё успешно установлено и сервис работает:

   sudo systemctl status rabbitmq‑server.service

   

3. Включим пользовательский веб‑интерфейс для управления брокером:

   sudo rabbitmq‑plugins enable rabbitmq_management

4. Добавим пользователя для входа на консоль и назначим ему права администратора:

   sudo rabbitmqctl add_user #user# #password#
   sudo rabbitmqctl set_user_tags #user# administrator

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

Большинство операций и настроек можно провести через web‑консоль, используя в строке поиска браузера: #ip‑адрес хоста#:15672 и авторизовавшись под ранее заданной учётной записью администратора, но дальше в статье мы будем оперировать командами через rabbitmqctl и rabbitmqadmin.

Особенности RabbitMQ

Главной особенностью RabbitMQ является гибкая маршрутизация сообщений с использованием обменников и очередей.

Обменники

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

• Direct: сообщение отправляется в очередь, если ключ маршрутизации (routing key) совпадает с ключом очереди. Это вариант для самой простой маршрутизации.

• Fanout: сообщение рассылается во все привязанные очереди без учёта ключей. Вариант для массовых рассылок.

• Topic: маршрутизация на основе шаблонов ключей маршрутизации. Подходит для сложной фильтрации сообщений.

• Headers: маршрутизация на основе заголовков сообщения. Подходит для маршрутизации на основе метаданных.

Они создаются и настраиваются командой declare exchange:

rabbitmqadmin declare exchange name=my‑exchange type=direct internal=true

При создании обменника можно указать дополнительные параметры, влияющие на его поведение:

1. Alternate Exchange: обеспечивает резервную маршрутизацию для неподходящих сообщений;

2. Internal Exchange: позволяет создавать сложные цепочки маршрутизации между обменниками;

3. TTL: управляет временем жизни сообщений;

4. Dead Letter Exchanges: обрабатывает «мёртвые» сообщения;

5. Max Length/Bytes: ограничивает размер очередей;

6. Custom Arguments: добавляет специфические параметры для ваших задач.

Пример создания обменника с дополнительными аргументами:

rabbitmqadmin declare exchange name=my-complex-exchange type=topic durable=true auto_delete=false arguments='{
  "alternate-exchange":"my-alternate-exchange",
  "x-message-ttl":60000,
  "x-dead-letter-exchange":"my-dlx-exchange",
  "x-max-length":1000,
  "x-max-length-bytes":10485760
}'

В итоге мы получаем устойчивый (не удаляется при перезагрузке и отсутствии очередей) обменник my‑complex‑exchange, который поддерживает альтернативный обменник, обменник мёртвых сообщений для обработки ошибок (DLX), с временем жизни сообщений 60 сек., ограничениями на длину (1000 сообщений) и размер (10 Мб).

Очереди

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

1. Для долгосрочного хранения используются durable-очереди, а сообщения в них помечают как «persistent» (сохранение на диск).

2. Для временных задач создают автоматически удаляемые (auto_delete) или исключительные (exclusive) очереди для одного соединения.

3. Для предотвращения потери данных настраивают Dead Letter Queues (DLQ) и используют TTL для автоматического удаления старых сообщений.

4. Для больших объёмов данных включают режим lazy очередей (x‑queue‑mode: lazy) и ограничивают размер очереди (x‑max‑length или x‑max‑length‑bytes).

5. Для приоритетной обработки настраивают приоритеты сообщений (x‑max‑priority).

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

Пример создания очереди с заданием нескольких параметров:

rabbitmqadmin declare queue name=my-queue durable=true arguments='{
  "x-message-ttl":60000, 
  "x-dead-letter-exchange":"my-dlx-exchange", 
  "x-dead-letter-routing-key":"dlq-routing-key", 
  "x-max-length":1000, 
  "x-queue-mode":"lazy"
}'

В результате мы получим очередь my‑queue с временем жизни сообщений 60 сек., отправкой ошибочных сообщений (DLX) с маршрутизацией (dlq‑routing‑key), ограничением на 1000 сообщений и сохранением всех сообщений на диск.

Преимущества

Кроме упомянутой выше маршрутизации RabbitMQ обладает ещё несколькими преимуществами: обеспечивает надёжную доставку сообщений благодаря контролю скорости потребления (QoS), подтверждению обработки и возможности отправки отклонённых сообщений в DLQ.

QoS

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

Пример контроля скорости потребления (задается на потребителе):

rabbitmqadmin set‑qos prefetch‑count=1

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

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

Этот механизм повышает надёжность системы и не позволяет потерять сообщения при сбоях получателя. Если потребитель подтвердил сообщение, то оно удаляется из очереди, а если просто отключился или отклонил с включённым параметром переотправки requeue=true, то RabbitMQ вернёт сообщение в очередь. При отклонении без параметра сообщение будет удалено или отправлено в DLQ.

DLQ

Если сообщение было отклонено всеми потребителями, превышено количество попыток обработки или просрочилось по TTL, то сообщение попадает в очередь Dead Letter Queue (DLQ), где уже вручную или специальным обработчиком этой очереди должно быть проанализировано и обработано.

Ещё хорошим преимуществом RabbitMQ является поддержка различных протоколов соединения (AMQP, STOMP, HTTP и так далее), что делает его гибким инструментом для работы в разнообразных технологических стеках и для различных типов приложений.

Плагины

Для RabbitMQ также есть много различных плагинов, которые расширяют его функциональность:

Плагины для интеграции и маршрутизации:

• Shovel Plugin: обеспечивает передачу сообщений между экземплярами RabbitMQ или другими брокерами, полезен для миграции данных и резервного копирования. Документация.

• Federation Plugin: объединяет несколько кластеров RabbitMQ в одну распределённую систему, позволяя маршрутизировать сообщения между ними. Документация.

Плагины для обеспечения доступности и отказоустойчивости:

• Quorum Queues Plugin: вводит более надёжную репликацию данных, обеспечивая высокую доступность и отказоустойчивость. Документация.

Плагины для протоколов и веб‑интеграции:

• STOMP Plugin: поддерживает протокол STOMP для клиентов, не использующих AMQP, расширяя интеграцию с веб‑приложениями. Документация.

• Web STOMP Plugin: реализует HTTP‑интерфейс для подключения через WebSocket и STOMP, позволяя использовать RabbitMQ в веб‑приложениях в реальном времени. Документация.

Плагин для мониторинга и диагностики:

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

Заключение

Как видите, самостоятельно установить одну ноду Kafka или RabbitMQ не сложно. Однако дальнейшая настройка и управление требуют времени и знаний. А если речь идёт о создании кластера, то задача становится ещё труднее.

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

Если хотите больше узнать об облачных сервисах Kafka и RabbitMQ или протестировать их, то напишите нам — специалисты Т1 Облако проконсультируют вас по интересующим вопросам, помогут подобрать подходящие конфигурации и оперативно подключить сервисы.

---

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