Введение
Создан заказ, затем создан ещё один заказ, а после этого первый заказ оплачен. Таблица текущих статусов может показывать только последнее состояние каждого заказа. Событие описывает то, что произошло, поэтому все три факта могут быть полезны как история изменений. Системам, реагирующим на новые заказы, нужен способ передавать эти факты между приложениями.
Apache Kafka хранит и доставляет записи событий. Производитель отправляет записи, брокер — это сервер, который получает и хранит их, топик задаёт имя потоку записей, а потребитель читает их. Kafka может связывать приложения, работающие в разное время или с разной скоростью. Другие инструменты обработки могут рассчитывать отчёты по этим событиям; в начале этого курса основное внимание уделяется надёжной доставке и отслеживанию позиции чтения.
Вы создадите топик, понаблюдаете за производителем и потребителем в разных терминалах и прочитаете те же события ещё раз после завершения первого потребителя. Подготовленная среда с одним брокером использует одну реплику; развёртывание кластера — отдельная тема. Для первых шести работ достаточно базовых навыков работы в терминале. В последующих работах с Python-клиентами потребуются функции, циклы, словари, JSON и запуск скриптов; подготовиться поможет курс Python. Знание Java, Hadoop, Hive и Spark, а также внешние учётные записи не требуются. Используйте сервис Terminal.
Создайте место для событий заказов
На этом шаге вы убедитесь, что подготовленный сервис Kafka отвечает, и создадите топик. Оставьте эту первую оболочку как терминал A. Перейдите в рабочий каталог:
## Terminal A
cd /home/labex/project
Клиенты командной строки уже установлены. Параметр --bootstrap-server задаёт адрес начального брокера, через который клиент получает сведения о кластере. Здесь localhost:9092 — подготовленный локальный сервис, для которого не нужна внешняя учётная запись. Выведите список его топиков, чтобы убедиться в ответе настоящего брокера:
## Terminal A
kafka-topics.sh --bootstrap-server localhost:9092 --list
В чистой учебной среде топиков приложений пока может не быть. Создайте orders-first. Топик содержит один или несколько разделов — упорядоченных журналов, в которых хранятся его записи. Начните с одного раздела, чтобы за первой последовательностью было легко следить. Параметр --replication-factor 1 подходит для среды с одним подготовленным брокером:
## Terminal A
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-first --partitions 1 --replication-factor 1
Просмотрите сведения о новом топике:
## Terminal A
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-first
Найдите имя топика, PartitionCount: 1 и ReplicationFactor: 1. Идентификаторы, номера лидеров и формат вывода могут различаться. Создание топика подготовило место хранения; оно ещё не отправило заказ и не установило дополнительный сервер.
Брокер продолжит работать, пока кратковременные клиенты производителей и потребителей запускаются и завершаются. Далее вы воспользуетесь этой независимостью, чтобы наблюдать поступление событий.
Понаблюдайте, как один клиент получает отправленное другим
На этом шаге вы оставите потребителя ждать в одном терминале и отправите три записи из другого. Сервис Terminal сохраняет оболочку с помощью tmux, менеджера сеансов терминала. Второе подключение из браузера изначально показывает тот же сохраняемый сеанс. Создайте для терминала B отдельный именованный сеанс оболочки, чтобы у клиентов были независимые командные строки.
В терминале A создайте отсоединённый сеанс. Параметр -d оставляет выбранным текущий терминал, -s задаёт имя нового сеанса, а -c выбирает его начальный каталог. Эта команда создаёт оболочку, а не ещё один брокер Kafka:
## Terminal A
tmux new-session -d -s kafka-b -c /home/labex/project
Откройте тот же сервис Terminal во второй вкладке браузера, не закрывая терминал A. Назовём новую вкладку терминалом B. Переключите это подключение на подготовленный сеанс:
## Terminal B
tmux switch-client -t kafka-b
После переключения один раз нажмите Ctrl-C, чтобы очистить ещё не отправленный ввод терминала, пока активна только оболочка. Проверьте имя сеанса:
## Terminal B
tmux display-message -p '#S'
Ожидается kafka-b. В терминале A остаётся исходный сеанс; обе оболочки имеют доступ к файлам одной и той же виртуальной машины. Сохраняйте эти роли при переключении вкладок браузера. Если именованный сеанс уже был создан в ходе этой работы, используйте его повторно вместо повторного создания.
В терминале B запустите потребителя следующей командой. Он ждёт, пока прочитает три записи, а затем автоматически завершается. Параметр --from-beginning запрашивает начало сохранённых данных для этого нового потребителя; --max-messages 3 ограничивает число сообщений, а --timeout-ms 30000 ограничивает ожидание бездействия. Конвейер передаёт стандартный вывод в tee, которая показывает его и сохраняет локальную копию:
## Terminal B
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-first --from-beginning --max-messages 3 --timeout-ms 30000 | tee /home/labex/project/output/received-events.txt
Оставьте терминал B в ожидании. Переключитесь в терминал A и отправьте пакет до истечения 30-секундного периода бездействия. Если время истекло до отправки, снова запустите ту же команду потребителя в терминале B, а затем вернитесь к отправке пакета. Если вы уже отправили события, повторно запустите только потребителя, чтобы просмотреть три сохранённые записи; повторная отправка пакета производителем добавит дубликаты:
## Terminal A
printf '%s\n' 'order-1001 created' 'order-1002 created' 'order-1001 paid' | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders-first
printf выводит каждое заключённое в кавычки текстовое значение с последующим переводом строки. Конвейер передаёт эти строки производителю; консольный производитель рассматривает каждую строку как одну запись события. Когда printf завершается, её поток ввода закрывается, поэтому производитель заканчивает ожидающие отправки и завершает работу. Вводить данные в интерактивную командную строку производителя не нужно.
Вернитесь в терминал B. В нём должны появиться три строки данных:
order-1001 created
order-1002 created
order-1001 paid
В терминале также могут появиться число обработанных сообщений или диагностические сообщения клиента. Сохранённый файл должен содержать три значения событий. Два факта об order-1001 — отдельные записи; событие paid не перезаписало предыдущее событие created.
Обычно после трёх сообщений потребитель возвращает управление командной строке. Если вы случайно запустили клиент без ограничения, Ctrl-C останавливает этот клиент в его терминале; брокер продолжает работать, а топик сохраняется. Здесь ограничения позволяют завершить обычный сценарий без ручного прерывания.
Прочитайте сохранённые события новым потребителем
На этом шаге вы проверите, удалились ли события после чтения. Первый потребитель уже завершил работу. Предположите, что увидит новый независимый потребитель, начинающий с начала сохранённых данных: те же три записи.
В терминале B запустите нового потребителя с ограничением и сохраните результат второго чтения отдельно:
## Terminal B
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-first --from-beginning --max-messages 3 --timeout-ms 30000 | tee /home/labex/project/output/replayed-events.txt
Ожидаются те же три значения в том же порядке в пределах этого единственного раздела. Вы не отправляли ещё один пакет. Просмотрите оба локальных результата из терминала A:
## Terminal A
cat /home/labex/project/output/received-events.txt
## Terminal A
cat /home/labex/project/output/replayed-events.txt
Оба файла должны содержать все три исходных события. Чтение и хранение данных независимы: чтение не удаляет запись из топика. В этом первом эксперименте используются новые консольные потребители. Позже вы узнаете, как именованная группа потребителей сохраняет позицию чтения и почему параметр начала чтения не отменяет уже зафиксированную допустимую позицию.
Kafka сохранила доступ к истории событий после завершения обоих кратковременных клиентов. Далее вы изучите свойства топика и увидите, как настройки хранения влияют на политику хранения независимо от позиции потребителей.
Итоги
Вы изучили роль Kafka в доставке событий, создали топик с одним разделом и понаблюдали, как отдельные клиенты производителя и потребителя обмениваются тремя событиями заказов. Новый потребитель прочитал ту же сохранённую историю после завершения первого, показав, что чтение и удаление — разные операции. Далее вы изучите свойства разделов, реплик и хранения топика.
