Очереди сообщений: пример
Ранее мы рассмотрели системы очередей сообщений в целом. Мы увидели, что сначала довольно долго применялись системы очередей сообщений общего назначения для межпроцессорного взаимодействия, а затем появились более специализированные очереди сообщений для систем клиент-сервер или архитектур веб-служб. Хотя существует несколько различных систем с уникальными конструкциями и функциями, мы рассмотрим систему, созданную с нуля для масштабируемости и эластичности: Apache Kafka.
Apache Kafka
Kafka обрабатывает сообщения от набора программ (так называемых производителей) и отправляет их в набор компьютеров (так называемых потребителей), которые могут быть заинтересованы в этих сообщениях. Производители публикуют сообщения в одну из тем в Kafka. Потребители могут подписаться на определенные темы, и тогда Kafka будет доставлять сообщения потребителям. Таким образом, Apache Kafka можно описать как распределенную систему обмена сообщениями по типу "публикация-подписка" с открытым исходным кодом.
Рис. 3. Кластер Kafka
Тема — это определяемая пользователем категория, в которую публикуются сообщения. Пример темы, которую можно встретить в рекламной компании, — AdClickEvents. Все потребители данных могут считывать данные из одной или нескольких тем. На внутреннем уровне каждая тема хранится как секционированный журнал фиксации, как показано на рис. 4. Следует отметить, что тема может состоять из нескольких разделов, а кластер Kafka может обрабатывать несколько тем.
Рис. 4. Очередь сообщений в Kafka
Формально каждый раздел представляет собой упорядоченную, неизменяемую последовательность сообщений, которые постоянно добавляются в журнал фиксации. Всем сообщениям в разделах присваивается порядковый код, который называется смещением. Он однозначно определяет каждое сообщение в разделе и не может использоваться для упорядочения сообщений в теме, находящихся в разных разделах.
Раздел в журнале служат для нескольких целей. Во-первых, они позволяют масштабировать журнал до размеров, превышающих доступное место на одном сервере. Каждый отдельный раздел должен соответствовать размеру серверов, на которых он хранится, однако тема может состоять из нескольких разделов и обрабатывать произвольное количество данных. Во-вторых, они действуют как единица параллелизма, что позволяет распределять отдельные разделы журнала между несколькими компьютерами. Производители не только контролируют тему сообщения, но и могут явно управлять разделом, в который отправляется сообщение, если это необходимо, используя функцию семантического секционирования (подобно функциям секционирования, которые используются в MapReduce). По умолчанию сообщения в определенной теме распределяются по разделам этой темы в порядке циклического перебора.
Кластер Kafka содержит все опубликованные сообщения (независимо от их потребления) в течение настраиваемого периода времени (по умолчанию 7 дней). Сообщения, которые хранились дольше этого периода, Kafka автоматически удаляет, чтобы освободить место для новых сообщений.
Kafka также отслеживает успехи каждого потребителя в чтении сообщений файла журнала по определенной теме — это называется смещением для потребителя. Большинство потребителей наращивают значение смещения линейно по мере того, как потребляют сообщения. Потребители управляют переменной смещения и при необходимости могут перемещаться вперед и назад для чтения более старых или более новых сообщений.
Такая комбинация функций делает потребители Kafka очень дешевыми. Их прибытие и выбытие никак не влияет на кластер или на других потребителей.
Гарантии, предоставляемые Apache Kafka
Kafka предоставляет набор высокоуровневых гарантий, на которые могут рассчитывать разработчики приложений.
- Сообщения, отправляемые производителем в определенный раздел темы, будут добавляться в том порядке, в котором они отправляются. Это значит, что, если сообщение M1 отправляется тем же производителем, что и сообщение M2, и сообщение M1 отправляется первым, то сообщение M1 получит меньшее смещение, чем M2 и будет отображаться в журнале раньше.
- Экземпляр потребителя видит сообщения в том порядке, в котором они хранятся в журнале.
- Для темы с коэффициентом репликации $N $ мы допускаем до $N-$1 отказов серверов без потери каких бы то ни было сообщений, зафиксированных в журнале.
Обратите внимание на, что гарантии доставки не очень строгие. Это значит, что в редких случаях потребители могут получать одно и то же сообщение дважды.
Архитектура Apache Kafka
Рис. 5. Архитектура Kafka
Серверы, которые передают сообщения от издателей (производителей) подписчикам (потребителям), называются брокерами Kafka. Брокеры Kafka отвечают за хранение и репликацию сообщений. Разделы каждой темы распределяются по брокерам, и каждый брокер хранит один или несколько разделов.
Брокеры упорядочиваются в децентрализованном порядке, так как фиксированного основного брокера нет. Для того чтобы брокеры могли достигать согласия о состоянии системы, используется Apache ZooKeeper. ZooKeeper предоставляет собой высокодоступную службу обеспечения консенсуса в виде API по типу файловой системы. ZooKeeper используется в Kafka для выполнения следующих задач.
- Обнаружение добавления и удаления брокеров и потребителей в системе
- Запуск перераспределения разделов при изменении количества брокеров или потребителей
- Поддержание связи потребления и отслеживание потребленных смещений по каждому разделу
Как уже говорилось, наименьшей единицей параллелизма в Kafka является раздел в теме. Это значит, что в любой момент времени все сообщения раздела потребляются одним потребителем. Это позволяет Kafka не выполнять дорогостоящую координацию между несколькими брокерами, как если бы она гарантировала сохранение порядка сообщений между разделами.
Для обеспечения отказоустойчивости разделы реплицируются между несколькими брокерами. Один из брокеров назначается лидером определенного раздела, и все операции чтения и записи для этого раздела по умолчанию отправляются в главную реплику. Сообщение считается зафиксированным только после того, как все реплики зафиксируют это сообщение в своем журнале. Потребителям направляются только зафиксированные сообщения. Производители могут выбирать блокировку до тех пор, пока сообщение не будет зафиксировано Kafka, или непрерывную потоковую передачу сообщений без блокировки. В Kafka используются несколько методов для ускорения процесса репликации журналов. Дополнительные сведения см. в документации по Apache Kafka.
Поскольку брокеры Kafka должны обрабатывать большие количества сообщений, Kafka отслеживает для каждого узла в кластере две характеристики "жизнеспособности":
- Каждый узел поддерживает сеанс с ZooKeeper с помощью механизма периодических сигналов.
- Каждый подчиненный объект должен реплицировать обновления главного объекта и не отставать от него слишком сильно. Задержку реплицирования Kafka можно настраивать для всех кластеров.
Взаимодействие с производителями
Производители могут отправлять сообщения в кластер Kafka с помощью API Kafka. Производители знают, какие темы и разделы настроены. Kafka и производитель обычно направляют сообщения в соответствующий брокер, который отвечает за обработку определенного раздела сообщения. API также позволяет создавать запросы метаданных. За счет этого производители могут запрашивать и размещать соответствующий брокер для темы и раздела. Как уже говорилось, секционирование раздела можно настроить, а также реализовать произвольное распределение нагрузки или секционирование темы с учетом содержимого.
Кроме того, производители, взаимодействующие с брокером Kafka, могут передавать сообщения асинхронно, а также запрашивать пакетную обработку, при которой сообщения накапливаются и отправляются пакетами. Их также можно настраивать, как количество сообщений на пакет, так и фиксированный предел задержки, что позволяет приложению жертвовать задержкой ради пропускной способности.
Взаимодействие с потребителями
Потребители Kafka создают запросы от брокеров для отдельных разделов, которые необходимо потребить. Потребитель может указать в запросе смещение в журнале Kafka и получить блок сообщений, начиная с указанной позиции. Потребители могут переходить к любому смещению и запрашивать сообщения, пока не истек период их хранения в кластере Kafka.
Варианты использования Kafka
Очередь обмена сообщениями: неудивительно, Kafka можно использовать в качестве замены для традиционных очередей обмена сообщениями, таких как ActiveMQ или RabbitMQ. Kafka имеет особое значение, поскольку эта платформа разработана с нуля и обеспечивает высокую доступность и масштабируемую доставку сообщений с настраиваемыми требованиями к задержке и пропускной способности.
Отслеживание действий веб-сайта: Kafka изначально был создан LinkedIn для создания конвейера действий пользователей и принятия решений в режиме реального времени для контента и размещения объявлений для пользователей LinkedIn. В этом сценарии темы могут определяться типом взаимодействия с пользователем (например, одна тема может быть предназначена для просмотров страницы и информации о прокрутке, другая — для поисковых запросов, а третья — для переходов пользователей по ссылкам). Различные серверные службы, такие как обработка и мониторинг активности пользователей в режиме реального времени, могут подписываться на соответствующие темы и обрабатывать потоки по мере их появления.
Агрегирование журналов: Kafka можно использовать для статистической обработки журналов из нескольких служб и сделать их доступными в центральном расположении для обработки. По сравнению с системами, ориентированными на ведение журналов, такими как Scribe или Flume, Kafka предлагает одинаково хорошую производительность и лучше гарантирует устойчивость за счет репликации и гораздо более низкой сквозной задержки.