Часть 1: Знакомство с Кафкой
Apache Kafka — это распределенная платформа потоковой передачи. Что именно это означает? Потоковая платформа имеет три ключевые особенности: Публикуйте потоки записей и подписывайтесь на них, подобно очередям сообщений или корпоративным системам обмена сообщениями. Сохраняйте потоки записей отказоустойчивым и устойчивым образом. Поток обработки, когда происходит запись.
Kafka обычно используется в двух широких категориях приложений:
1. Создайте конвейеры потоковой передачи данных в реальном времени для надежного сбора данных между системами или приложениями.
2. Создавайте приложения для потоковой передачи в реальном времени, которые преобразуют потоки данных или реагируют на них.
Чтобы понять, как это делает Kafka, давайте рассмотрим возможности Kafka снизу вверх.
Сначала несколько понятий:
Kafka работает как кластер на одном или нескольких серверах, которые могут охватывать несколько центров обработки данных.
Кластеры Kafka хранят потоки записей в категориях, называемых темами. Каждая запись состоит из ключа, значения и метки времени.
Kafka имеет четыре основных API:
Producer API (Producer API) позволяет приложению публиковать поток записей в одну или несколько тем Kafka.
Потребительский API (Consumer API) позволяет приложению подписываться на одну или несколько тем и обрабатывать поток создаваемых для них записей.
Streams API (Streams API) позволяет приложению действовать как потоковый процессор, потребляя входные потоки из одной или нескольких тем и создавая выходные потоки для одной или нескольких выходных тем, эффективно преобразовывая входные потоки в выходные потоки.
API Connector позволяет создавать и запускать многоразовых производителей или потребителей для подключения тем Kafka к существующим приложениям или системам данных. Например, коннектор к реляционной базе данных может фиксировать каждое изменение в таблице.
Topics and Logs
Давайте сначала погрузимся в основную абстракцию Kafka предоставляет темы для потока записей.
Тема – это категория или название канала, в котором опубликована запись. Темы в Kafka всегда многопользовательские, то есть в теме может быть один или несколько пользователей, подписанных на эти данные.
Для каждой темы кластер Kafka ведет секционированный журнал, который выглядит следующим образом:
Каждый раздел представляет собой упорядоченную неизменяемую последовательность записей, которые постоянно добавляются в структурированный журнал фиксации. Записям в каждом разделе назначается последовательный идентификационный номер, называемый смещением, который однозначно идентифицирует каждую запись в разделе.
Все опубликованные записи сохраняются в кластере Kafka независимо от того, использовались ли они с использованием настраиваемого периода хранения. Например, если политика хранения установлена на два дня, то через два дня после публикации запись будет доступна для использования, после чего она будет удалена для освобождения места. Производительность Kafka фактически не зависит от размера данных, поэтому хранение данных в течение длительного периода времени не является проблемой.
По сути, уникальные метаданные, зарезервированные на основе каждого потребителя, — это смещение или местоположение пользователя в журнале. Это смещение контролируется потребителями: в нормальных условиях потребитель будет линейно сдвигать свое смещение при чтении записи, но на самом деле, поскольку местоположение контролируется потребителем, он может нажимать и смотреть в любом порядке. Например, потребители могут сбросить старое смещение, чтобы повторно обработать прошлые данные, или пропустить самые последние записи и начать потребление «сейчас».
Эта комбинированная функция означает, что потребители Kafka очень дешевы, они могут приходить и уходить без особого влияния на кластер или других потребителей. Например, вы можете использовать наш инструмент командной строки, чтобы «отслеживать» содержимое любой темы, не изменяя содержимое, потребляемое какими-либо существующими потребителями.
Разделы в журнале служат различным целям. Во-первых, они позволяют журналам превышать размер, подходящий для одного сервера. Каждый отдельный раздел должен соответствовать серверу, на котором он размещен, но тема может иметь много разделов, поэтому она может обрабатывать любой объем данных. Во-вторых, на данный момент они больше похожи на параллельные единицы.
Distribution
Разделы журнала распределены по серверам в кластере Kafka, при этом каждый сервер обрабатывает данные и запрашивает общий раздел. Каждый раздел реплицируется на несколько настраиваемых серверов для обеспечения отказоустойчивости.
В каждом разделе есть один сервер, который действует как «лидер», и ноль или более серверов, которые действуют как «последователи». Лидер обрабатывает все запросы на чтение и запись в раздел, а последователи пассивно копируют лидера. Если лидер терпит неудачу, один из последователей автоматически становится новым лидером. Каждый сервер выступает в качестве ведущего для одних разделов и подчиненного для других, поэтому нагрузка внутри кластера хорошо сбалансирована.
Geo-Replication
Kafka MirrorMaker обеспечивает поддержку георепликации для вашего кластера. Используя зеркальные машины, сообщения реплицируются в нескольких центрах обработки данных или облачных регионах. Это резервное копирование и восстановление можно использовать в сценариях «активный/пассивный» или в сценариях «активный/активный», чтобы приблизить данные к пользователю или удовлетворить требования к расположению данных.
Producers
Производители публикуют данные в теме по своему выбору. Производитель отвечает за выбор записей раздела в теме для назначения. Это можно сделать в циклическом режиме, просто чтобы сбалансировать нагрузку, или это можно сделать, разделив функцию в соответствии с некоторой семантикой (например, на основе определенных ключей в записи).
Consumers
Потребители помечают себя именем группы потребителей, и каждая запись, опубликованная в теме, доставляется одному экземпляру потребителя в каждой подписавшейся группе потребителей. Экземпляры-потребители могут находиться в отдельных процессах или на отдельных машинах.
Если все экземпляры-потребители имеют одну и ту же группу потребителей, записи эффективно распределяют нагрузку между экземплярами-потребителями.
Если все экземпляры потребителей имеют разные группы потребителей, то каждая запись будет транслироваться всем процессам-потребителям.
Кластер Kafka с двумя серверами содержит четыре раздела (от P0 до P3) с двумя группами потребителей. Группа потребителей A имеет два экземпляра потребителей, а группа B — четыре.
Однако чаще мы обнаруживаем, что темы имеют небольшое количество групп потребителей, каждая из которых имеет «логического пользователя». Каждая группа состоит из множества экземпляров-потребителей для обеспечения масштабируемости и отказоустойчивости. Это просто семантика публикации-подписки, где подписчик — это набор потребителей, а не отдельный процесс.
Потребление реализовано в Kafka путем разделения разделов в журнале на экземпляры-потребители таким образом, что каждый экземпляр является единственным потребителем «справедливой доли» разделов в любой момент времени. Процесс сохранения членства в этой группе динамически обрабатывается протоколом Kafka. Если к группе присоединятся новые экземпляры, они возьмут на себя некоторые разделы от других членов группы; если экземпляр умрет, его разделы будут распределены между оставшимися экземплярами.
Кафка записал только общий порядок разделов, а не между разными разделами по теме. Ключ сортировки для каждого раздела Объем привязки данных раздела достаточен для большинства приложений. Однако, если вам нужно отсортировать общее количество записей, это может быть достигнуто только с одним разделом темы, но это будет означать, что каждая группа потребителей имеет только один процесс-потребитель.
Мультиарендная технология
Kafka можно развернуть как многопользовательское решение. Мультиарендность достигается путем настройки тем, которые могут создавать или потреблять данные. Существует также оперативная поддержка квот. Администраторы могут определять и применять квоты для запросов, чтобы контролировать прокси-ресурсы, используемые клиентами.
гарантировать
На высоком уровне Kafka предоставляет следующие гарантии:
Сообщения, отправленные производителем в определенный раздел темы, будут добавлены в том порядке, в котором они были отправлены. То есть, если запись M1 была отправлена тем же производителем, что и запись M2, и M1 была отправлена первой, то M1 будет иметь меньшее смещение, чем M2, и появиться в журнале раньше.
Экземпляр пользователя просматривает записи в том порядке, в котором они хранятся в журнале.
Для темы с коэффициентом репликации N допустимо до N-1 сбоев сервера без потери каких-либо записей, зафиксированных в журнале.
Kafka может гарантировать, что сообщения в одном и том же разделе упорядочены, то есть производители отправляют сообщения в определенном порядке, брокер записывает их в раздел в этом порядке, и потребители читают их в том же порядке.
###### Kafka作为消息传递系统
Чем отличается концепция потоков Kafka от традиционных корпоративных систем обмена сообщениями?
Традиционно существует два режима обмена сообщениями: очередь и публикация-подписка. В очереди пул потребителей может читать с сервера, и каждая запись отправляется одному из них; в публикации-подписке записи рассылаются всем потребителям. Оба режима имеют свои преимущества и недостатки. Преимущество очередей в том, что они позволяют разделить обработку данных между несколькими экземплярами потребителей, что позволяет масштабировать обработку. К сожалению, как только процесс прочитал свои данные, очередь больше не является многопользовательской. Публикация-подписка позволяет передавать данные нескольким процессам, но, поскольку каждое сообщение передается каждому подписчику, масштабирование невозможно.
Концепция Кафки о потребительских группах заключает в себе эти две концепции. Подобно очередям, группы потребителей позволяют разделить обработку на несколько процессов (членов группы потребителей). Как и публикация-подписка, Kafka позволяет рассылать сообщения нескольким группам пользователей.
Преимущество модели Kafka в том, что каждая тема имеет эти свойства, она масштабирует обработку, а также является многопользовательской, нет необходимости выбирать ту или иную.
Kafka имеет более сильные гарантии упорядочения, чем традиционные системы обмена сообщениями.
Традиционные очереди сохраняют порядок записей на сервере, и если несколько пользователей используют очередь из очереди, сервер распределяет записи в том порядке, в котором они хранятся. Однако, хотя сервер распределяет записи по порядку, записи доставляются потребителям асинхронно, поэтому они могут поступать к разным потребителям не по порядку. Фактически это означает, что порядок записей теряется при параллельном потреблении. Системы обмена сообщениями часто обходят это с помощью концепции «эксклюзивного потребителя», которая позволяет только одному процессу потреблять из очереди, но, конечно, это означает отсутствие параллелизма в обработке.
Кафка лучше. Kafka может обеспечить гарантию сортировки и балансировку нагрузки в пуле процессов-потребителей с помощью концепции параллельных разделов в субъекте. Это достигается путем назначения разделов в субъекте потребителям в группе потребителей, так что каждый раздел полностью потребляется одним из потребителей в этой группе. Благодаря этому мы гарантируем, что потребители являются единственными читателями раздела и потребляют данные по порядку. Поскольку существует много разделов, это по-прежнему уравновешивает нагрузку многих экземпляров-потребителей. Обратите внимание, что экземпляры потребителей в группе потребителей не могут быть больше, чем разделы.
Kafka как система хранения
Любые очереди сообщений, которые позволяют публиковать сообщения, и разделение между ними эффективно действуют как система хранения для сообщений в пути. Kafka отличается тем, что это очень хорошая система хранения.
Данные, записываемые в Kafka, записываются на диск и реплицируются для обеспечения отказоустойчивости. Kafka позволяет производителям ждать подтверждений, так что запись считается завершенной до тех пор, пока она не будет полностью реплицирована, и гарантированно сохранится, даже если сервер не сможет выполнить запись.
Структура на диске Kafka хорошо использует масштабирование Kafka будет работать одинаково независимо от того, есть ли у вас 50 КБ или 50 ТБ постоянных данных на сервере.
В результате серьезного отношения к хранилищу и предоставления клиентам возможности контролировать, где они читают, вы можете думать о Kafka как о выделенной распределенной файловой системе, предназначенной для высокопроизводительного хранения журнала фиксации, репликации и распространения с малой задержкой.
Kafka для потоковой обработки
Недостаточно читать, записывать и хранить потоки данных, его цель — добиться обработки конвекции в реальном времени.
В Kafka потоковый процессор — это все, что берет непрерывный поток данных из входной темы, выполняет некоторую обработку этих входных данных и создает непрерывный поток данных для выходной темы.
Например, приложение для розничной торговли может получать входной поток продаж и отгрузок и выводить поток повторных заказов и корректировок цен, рассчитанный на основе этих данных.
Простая обработка может выполняться напрямую с помощью API-интерфейсов производителя и потребителя. Однако для более сложных преобразований Kafka предоставляет полностью интегрированный потоковый API. Это позволяет создавать нетривиальные приложения для обработки, которые могут вычислять агрегаты в потоках или объединять потоки вместе.
Этот инструмент помогает решить проблемы, с которыми сталкиваются такие приложения: обработка неупорядоченных данных, повторная обработка ввода, изменения кода, выполнение вычислений состояния и т. д.
Streaming API построен на основных примитивах, предоставляемых Kafka: он использует API производителя и потребителя для ввода, Kafka для хранения состояния и один и тот же групповой механизм для обеспечения отказоустойчивости для экземпляров потокового процессора.
собрать кусочки вместе
Такое сочетание обмена сообщениями, хранения и потоковой обработки может показаться необычным, но оно имеет решающее значение для роли Kafka как потоковой платформы.
Распределенные файловые системы, такие как HDFS, позволяют хранить статические файлы для пакетной обработки. По сути, такая система позволяет хранить и обрабатывать исторические данные из прошлого.
Традиционные корпоративные системы обмена сообщениями позволяют обрабатывать будущие сообщения, поступающие после подписки. Приложения, построенные таким образом, обрабатывают будущие данные по мере их поступления.
Kafka сочетает в себе эти две возможности, и это сочетание имеет решающее значение для Kafka как платформы приложений потокового мультимедиа, а также конвейера потоковых данных.
Сочетая хранилище и подписки с малой задержкой, потоковые приложения могут одинаково обрабатывать прошлые и будущие данные. То есть одно приложение может обрабатывать исторические, сохраненные данные, а не заканчивать работу при достижении последней записи, оно может продолжать обработку по мере поступления будущих данных. Это общая концепция потоковой обработки, включая пакетные приложения и приложения, управляемые сообщениями.
Аналогичным образом, для потоковых конвейеров данных подписка на события в реальном времени позволяет использовать Kafka для конвейеров с очень малой задержкой, однако возможность надежного хранения данных делает его полезным для критически важных данных, где должны быть гарантированы доставка или интеграция данных. n имеет автономную систему, которая загружает данные только периодически или может поддерживаться в течение длительного периода времени. Средства потоковой обработки позволяют преобразовывать данные по мере их поступления.
Часть II: Передовая технология Kafka
Резюме производителя
Основные этапы отправки сообщений в Kafka Мы начинаем с создания объекта ProducerRecord. Объект ProducerRecord должен содержать целевую тему и содержимое для отправки. Мы также можем указать ключевые разделы. При отправке объекта ProducerRecord производитель должен сначала сериализовать объекты ключа и значения в массивы байтов, чтобы они передавались по сети. Затем данные передаются разделителю. Если раздел был ранее указан в объекте ProducerRecord, разделитель ничего не сделает и вернет указанный раздел напрямую. Если раздел не указан, разделитель будет основан на объекте ProducerRecord. ключ объекта для выбора раздела. После того, как раздел выбран, производитель знает, в какую тему и раздел отправить сообщение. Далее эта запись добавляется в пакет записей, и все сообщения в этом пакете будут отправляться в ту же тему и партицию, за отправку этих пакетов записей соответствующим брокерам отвечает отдельный поток. Сервер возвращает ответ, когда он получает эти сообщения. Если сообщение успешно записано в Kafka, он возвращает объект RecordMetaData, который содержит информацию о теме и разделе, а также смещение записи в разделе. Если запись не удалась, он возвращает Ошибка. Производитель попытается повторно отправить сообщение после получения ошибки, и если он все еще терпит неудачу через несколько раз, он вернет сообщение об ошибке.
Создать производителя Kafka
Чтобы написать сообщение в Kafka, вы должны сначала создать объект производителя и установить некоторые свойства.Производитель Kafka имеет 3 обязательных свойства (без объяснения, что это значит)
bootstrap.серверы key.serializer value.serializer Другие свойства по умолчанию: акки буфер.память сжатие.тип повторяет размер партии задерживаться.мс ID клиента max.in.flight.requests.oer.connection timeout.ms, request.timeout.ms и metadata.fetch.timeout.ms макс.блок.мс макс.запрос.размер resive.buffer.bytes и send.buffer.bytes
В следующем фрагменте кода показано, как создать нового производителя, здесь указаны только необходимые свойства, остальные используют настройки по умолчанию.
private Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers","broker1:9092,broker2:9092");
kafkaProps.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
producer = new KafkaProducer<String, String>(kafkaProps);
Основным значением вышеуказанного фрагмента кода является создание новых объектов свойств в первую очередь Поскольку мы намереваемся определить ключи и значения как типы строки, используйте встроенные строкириализатора Здесь мы создаем новый объект производителя, установите соответствующий тип для ключа и значения и пропустить его объект свойств
После создания экземпляра объекта производителя вы можете начать отправлять сообщения. Существует три основных способа отправки сообщений:
огонь и фальсификатор:Мы отправляем сообщение на сервер, но не важно, нормально ли оно приходит.Большую часть времени сообщение будет приходить нормально, потому что Kafka высокодоступна, и производитель автоматически попытается отправить повторно, но иногда при использовании этого метода также теряются некоторые сообщенияСинхронная отправка send():Мы используем метод send() для отправки сообщения, он вернет объект Future, вызовет метод get() для ожидания, и тогда вы сможете узнать, успешно ли отправлено сообщение.Асинхронная отправка send():Мы используем send() для отправки сообщения и указываем функцию обратного вызова, которую сервер вызывает при возврате ответа.
В приведенных выше примерах используется один поток, но на самом деле производитель может использовать несколько потоков для отправки сообщений.В начале может использоваться один потребитель и один поток.Если требуется более высокая пропускная способность, количество производителей может быть другим Увеличьте количество потоков, если это изменится. Если этого недостаточно, увеличьте количество производителей
Отправить сообщение Кафке
Самый простой способ отправить сообщение выглядит следующим образом
ProducerRecord<String, String> record = new ProducerRecord<>("CustomerCountry":"Precision Products","France");
try{
proucer.send(record);
} catch (Exception e){
e.printStackTrance();
}
Метод производителя send() принимает объект ProducerRecord в качестве параметра, поэтому нам нужно сначала создать объект ProducerRecord.ProducerRecord имеет несколько конструкторов, здесь используется только один из них, ему нужно имя целевой темы, ключ и значение объекты для отправки, они оба являются строками.Свежесть объектов ключа и значения должна соответствовать объекту-производителю сериализатора.
Мы используем метод производителя send() для отправки объекта ProducerRecord.Как видно из схемы архитектуры производителя, сообщение сначала помещается в буфер, а затем отправляется на сервер с помощью отдельного потока.Метод send() будет вернуть сообщение, содержащее объект Future RecordMetadata, но мы проигнорируем возвращаемое значение, поэтому невозможно узнать, было ли сообщение отправлено успешно.Если вам не важен результат отправки, вы можете использовать этот метод отправки. Например, записывать журналы сообщений Twitter или менее важные журналы приложений.
Отправить синхронно
Самый простой способ отправить сообщение синхронно выглядит следующим образом.
ProducerRecord<String, String> record = new ProducerRecord<>("CustomerCountry","Precision Products","France");
try{
producer.send(record).get();
} catch (Exception e){
e.printStackTrance();
}
Здесь метод производителя.send() сначала возвращает объект Future, а затем вызывает метод get() объекта Future, чтобы дождаться ответа Kafka. Если сервер возвращает ошибку, метод get() выдает исключение. Если нет возникает ошибка, мы получим объект RecordMetadata, который можно использовать для получения смещения сообщения Если перед отправкой данных или в процессе отправки возникает какая-либо ошибка, например, брокер возвращает исключение, которое не разрешает повторную передачу сообщения или количество повторных передач превышено, то будет выброшено исключение, мы просто печатаем исключение. информация выходит
Асинхронная отправка
Предположим, что сообщению требуется 10 мс, чтобы пройти туда и обратно между приложением и кластером Kafka.Если вы ждете ответа после отправки каждого сообщения, то для отправки 100 сообщений требуется 1 секунда, но если вы просто отправляете сообщения, не дожидаясь ответа, затем отправить 100 сообщенийЭто займет намного меньше времени.Большую часть времени нам не нужно ждать ответа, хотя Кафка отправит обратно целевую тему, информацию о разделе и смещение сообщения, но это не обязательно для приложения на стороне отправки.Однако при сбое отправки сообщения нам нужно создать исключение, записать журнал ошибок или записать сообщение в файл сообщений об ошибках.
Чтобы обрабатывать исключения при асинхронной отправке сообщений, производитель предоставляет поддержку обратного вызова.Ниже приведен пример использования обратных вызовов.
private class DemoProducerCallback implements Callback {
@Override
public void onCompletion(RecordMetadata recordMetadata,Exception e){
if (e != null){
e.prinStackTrace();
}
}
}
ProducerRecord<String, String> record = new ProducerRecord<>("CustomerCountry","Biomedical Materials","USA");
producer.send(record,new DemoProducerCallback());
Чтобы использовать обратные вызовы, вам нужен класс, реализующий интерфейс org.apach.kafka.clients.producer.Callback, который имеет только один метод onCompletion. Если Kafka возвращает ошибку, метод onCompletion выдает исключение, отличное от null. Передайте объект обратного вызова при отправке сообщения
Потребитель (KafkaConsumer)
Прежде чем понять, как читать сообщения от Kafka, давайте сначала разберемся с концепцией потребителей и групп потребителей. Предположим, у нас есть приложение, которое хочет читать сообщения из темы Kafka и проверять эти сообщения, а затем сохранять их.Приложению необходимо создать объект-потребитель, подписаться на тему и начать получать сообщения, затем проверять сообщения и сохранять результаты. , после этого Какое-то время производитель заставляет тему писать сообщение быстрее, чем приложение может проверить данные. Что мне делать в это время? Если для обработки сообщения используется только один потребитель, приложение никогда не успевает со скоростью генерации сообщений. В настоящее время необходимо: Так же, как несколько производителей могут писать сообщения в одну и ту же тему, нам также необходимо использовать несколько потребителей для чтения сообщений из одной и той же темы и распространения сообщений. Потребители Kafka принадлежат потребителю группы,Потребители в группе подписываются на одну и ту же тему, каждый потребитель получает сообщения из части раздела темы, сообщение темы в разделе будет подписано разными группами потребителей, и на сообщение может подписаться только каждая группа потребителей. получает потребитель.
Группы потребителей и ребалансировка разделов
Потребители в группе читают тематический раздел вместе.Когда новый потребитель присоединяется к группе, он читает сообщения, первоначально прочитанные другими потребителями.Когда потребитель закрывается или падает, он просто выходит из группы, раздел, изначально прочитанный им, будет читаются другими потребителями в группе. При изменении темы, например, когда администратор добавляет новый раздел, происходит перераспределение раздела. Владелец раздела меняется с одного. Действие передачи потребителя другому потребителю называетсяперебалансироватьКак программа запускает ребаланс? Потребители сохраняют свою принадлежность к группе и право собственности на разделы, отправляя контрольные сигналы брокеру координатора группы. Пока потребитель может отправлять контрольные сигналы через регулярные промежутки времени, он считается активным. Чтение сообщений из разделов. Потребители будут отправлять контрольные сигналы при опросе сообщения или коммит.
Создайте потребителя Kafka
{
Properties props = new Properties();
props.put("bootstrap.servers","broker1:9092,broker2:9092");
props.put("group.id","CountryCounter");
props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);
}
Подписывайтесь на темы
consumer.subscribe(Collection.singletonList("customerCountries"));//主题名"customerCountries"
Конфигурация потребителя
fetch.min.bytes fetch.max.wait.ms max.partition.fetch.bytes session.timeout.ms auto.offset.reset enable.auto.commit partition.assignment.strategy client.id max.poll.records receive.buffer.bytes/send.buffer.bytes
Здесь представлено основное содержание Кафки. Если вы хотите узнать о них больше, этого далеко не достаточно. Здесь вы можете порекомендовать вам несколько книг.
Если вы хотите иметь глубокое представление о Кафке в целом, вы можете прочитать > обязательное чтение
Второй — «Kafka Technology Insider». После прочтения этих двух книг все будет в порядке.
Наконец, вы также можете прочитать > читать не рекомендуется
Конечно, официальный сайт лучше всего подходит для чтения англоязычных документов.