Облачная платформаAdvanced

Начало работы с Kafka для производства и потребления сообщений

Язык статьи: Русский
Показать оригинал
Страница переведена автоматически и может содержать неточности. Рекомендуем сверяться с английской версией.

В этом разделе рассматривается пример создания инстанса Kafka (ciphertext access и SASL_SSL) и доступа к нему на клиенте (private network, внутри виртуальной частной облачной сети (VPC)) для производства и потребления сообщений, чтобы быстро начать работу с Distributed Message Service (DMS) for Kafka.

Figure 1 Процедура использования DMS for Kafka
  1. Шаг 1: Подготовка

    Инстанс Kafka работает в виртуальной частной облачной сети (VPC). Перед созданием инстанса Kafka убедитесь, что VPC доступна.

    После создания инстанса Kafka загрузите и установите открытый клиент Kafka на ваш ECS перед производством и потреблением сообщений.
  2. Шаг 2: Создание инстанса Kafka

    При создании инстанса Kafka вы можете выбрать спецификацию и количество, а также включить ciphertext access и SASL_SSL.

    При подключении к инстансу Kafka с включённым SASL_SSL используется SASL для аутентификации. Данные шифруются SSL‑сертификатами для передачи с высоким уровнем безопасности.
  3. Шаг 3: Создание топика

    Топики хранят сообщения, созданные продюсерами и получаемые потребителями.

    В этом разделе используется пример создания топика в консоли.
  4. Шаг 4: Подключение к инстансу Kafka для производства и потребления сообщений

    Перед подключением к инстансу Kafka с включённым SASL_SSL загрузите сертификат и настройте соединение в файле конфигурации клиента.

Шаг 1: Подготовка

  1. Предоставьте разрешения для экземпляра Kafka.

    Для обеспечения детального управления вашими облачными ресурсами создайте группы пользователей и пользователей Identity and Access Management (IAM) и предоставьте указанным пользователям соответствующие разрешения. Для получения дополнительной информации см. Creating an IAM User and Granting DMS for Kafka Permissions.

  2. Создайте VPC и подсеть.

    Перед созданием экземпляра Kafka убедитесь, что VPC и подсеть доступны. Подробную информацию о том, как создать VPC и подсеть, см. Creating a VPC.

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

  3. Создайте группу безопасности и добавьте правила группы безопасности.

    Перед созданием экземпляра Kafka убедитесь, что группа безопасности доступна. Подробную информацию о том, как создать группу безопасности, см. Creating a Security Group.

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

    Чтобы подключиться к экземплярам Kafka, добавьте правила группы безопасности, описанные в Table 1. Другие правила могут быть добавлены в соответствии с требованиями площадки.

    Table 1 правила группы безопасности

    Направление

    Протокол

    Порт

    Исходный адрес

    Описание

    Inbound

    TCP

    9093

    0.0.0.0/0

    Доступ к экземпляру Kafka через частную сеть внутри VPC (в зашифрованном виде)
    Note

    После создания группы безопасности у неё есть правило входящего трафика по умолчанию, позволяющее взаимодействовать ECS внутри группы безопасности, и правило исходящего трафика по умолчанию, позволяющее весь исходящий трафик. Если вы получаете доступ к вашему экземпляру Kafka через частную сеть внутри VPC, добавлять правила, описанные в Table 1, не требуется.

  4. Создайте клиент для производства и потребления сообщений.

    В этом разделе в качестве клиента используется Linux Elastic Cloud Server (ECS). Перед созданием экземпляра Kafka создайте ECS с EIP, установите JDK, настройте переменные окружения и загрузите клиент Kafka с открытым исходным кодом.

    1. Войдите в консоль, нажмите в левом верхнем углу, нажмите Elastic Cloud Server в разделе Computing, а затем создайте ECS.

      Подробную информацию о создании ECS см. в Creating an ECS. Если у вас уже есть доступный ECS, пропустите этот шаг.
    2. Войдите в ECS от имени пользователя root.
    3. Установите Java JDK и настройте переменные окружения JAVA_HOME и PATH.
      1. Скачайте JDK.
        Note

        Используйте Oracle JDK вместо JDK по умолчанию в ECS (например, OpenJDK), поскольку JDK по умолчанию в ECS может быть неподходящим. Получите Oracle JDK версии 1.8.111 или новее с Oracle's official website.
      2. Распакуйте JDK.
        tar -zxvf jdk-8u321-linux-x64.tar.gz

        Замените jdk-8u321-linux-x64.tar.gz на вашу версию JDK.
      3. Open the .bash_profile file.
        vim ~/.bash_profile
      4. Нажмите i и добавьте следующее содержимое.
        export JAVA_HOME=/root/jdk1.8.0_321 
        export PATH=$JAVA_HOME/bin:$PATH

        Измените /root/jdk1.8.0_321 на путь, где вы установили JDK.
      5. Нажмите Esc. Введите следующую строку и нажмите Enter. Сохраните файл .bash_profile и выйдите.
        :wq
      6. Выполните следующую команду, чтобы изменения вступили в силу:
        source .bash_profile
      7. Проверьте, установлен ли JDK.
        java -version

        Если возвращается следующее сообщение, JDK установлен.

        java version "1.8.0_321"

    4. Скачайте открытый Kafka client.
      wget https://archive.apache.org/dist/kafka/2.7.2/kafka_2.12-2.7.2.tgz
    5. Выполните следующую команду, чтобы распаковать пакет:
      tar -zxf kafka_2.12-2.7.2.tgz

Шаг 2: Создать Kafka Instance

  1. Войдите в консоль Kafka, затем нажмите Buy Kafka Instance в правом верхнем углу страницы.
  2. Укажите базовые параметры инстанса. Подробнее см. Table 2.

    Table 2 Базовые параметры инстанса

    Параметр

    Описание

    Режим биллинга

    Выберите Pay-per-use, который является постоплатным режимом. Вы можете платить после использования сервиса, и вам будет выставлен счет за продолжительность использования. Плата рассчитывается в секундах и выставляется почасово.

    Region

    DMS for Kafka в разных регионах не могут взаимодействовать друг с другом через интранет. Выберите ближайшее расположение для низкой задержки и быстрого доступа.

    Выберите RU-Moscow.

    Project

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

    Выберите RU-Moscow (по умолчанию).

    AZ

    AZ — это физический регион, где ресурсы используют независимое электропитание и сети. AZ физически изолированы, но соединены через внутреннюю сеть.

    Выберите AZ1, AZ2 и AZ3.

    Instance Name

    Вы можете задать имя, соответствующее правилам: 4–64 символа; начинается с буквы; может содержать только буквы, цифры, дефисы (-) и подчёркивания (_).

    Введите kafka-test.

    Enterprise Project

    Этот параметр предназначен для корпоративных пользователей. Enterprise project управляет ресурсами проекта в группах. Enterprise projects логически изолированы.

    Выберите default.

    Version

    Версия Kafka. Не может быть изменена после создания инстанса.

    Выберите 2.7.

    Broker Flavor

    Выберите broker flavor по требованию.

    Выберите kafka.2u4g.cluster.

    Brokers

    Укажите количество брокеров по требованию.

    Введите 3.

    Storage Space per Broker

    Выберите тип диска и укажите размер диска по требованию. Примечание: Вместимость диска может быть установлена только целым кратным 100.

    Общий объём хранения = Storage space per broker × Broker quantity. После создания инстанса изменить тип диска нельзя.

    Выберите Ultra-high I/O и введите 100.

    Capacity Threshold Policy

    Выберите Automatically delete: Когда диск достигает порога заполнения (95 %), сообщения могут продолжать создаваться и потребляться, но первые 10 % сообщений будут удалены для обеспечения достаточного свободного места. Используйте эту политику для сервисов, не допускающих прерываний. Однако данные могут быть потеряны.

  3. Настройте сеть instance. Для получения подробной информации см. Table 3.

    Table 3 Настройка сети instance

    Параметр

    Описание

    VPC

    VPC и подсеть нельзя изменить после создания instance.

    Выберите VPC и подсеть, подготовленные в 2.

    Security Group

    Выберите Security Group, подготовленную в 3.

  4. Установите режим доступа instance. Для получения подробной информации см. Table 4.

    Table 4 Настройка режима доступа instance

    Параметр

    Подпараметр

    Описание

    Private Network Access

    Plaintext Access

    Skip it.

    Ciphertext Access

    Когда этот параметр включён, требуется аутентификация SASL при подключении клиента к экземпляру Kafka.

    1. Ciphertext Access включен.
    2. SASL_SSL выбран. Username и Password можно задать. Введите "test" в качестве имени пользователя. Имя пользователя нельзя изменить после включения Ciphertext Access.

      Имя пользователя должно содержать от 4 до 64 символов, начинаться с буквы и включать только буквы, цифры, дефисы (-) и подчёркивания (_)

      Пароль должен соответствовать следующим требованиям:
      • Содержит от 8 до 32 символов.
      • Содержит как минимум три типа следующих символов: заглавные буквы, строчные буквы, цифры и специальные символы `~!@#$%^&*()-_=+\|[{}];:'",<.>? и пробелы, и не может начинаться с дефиса (-).
      • Не может быть именем пользователя, написанным вперёд или назад.
    3. SASL/PLAIN включен.

    Private IP Addresses

    Выберите Auto: Система автоматически назначает IP-адреса из подсети.

    Public Network Access

    -

    Пропустить.

  5. Пропустить Advanced Settings.
  6. Нажмите Buy.
  7. Подтвердите информацию об instance и затем отправьте запрос.
  8. Нажмите Back to DMS for Kafka List и проверьте, создан ли instance.

    Создание instance занимает от 3 до 15 минут. В течение этого периода статус instance — Creating.
    • Если instance успешно создан, его статус меняется на Running.
    • If the instance is in the Failed state, delete it and try creating another one. If the instance creation fails again, contact customer service.
      Note

      Instances, которые не удалось создать, не занимают другие ресурсы.

  9. После создания instance нажмите его имя, чтобы перейти на instance details page.
  10. In the Connection area, hover over the connection domain name to view and record the IP address.

    Figure 2 Адреса Kafka instance (частная сеть) для доступа внутри VPC

Шаг 3: Создать Topic

  1. On the Kafka Instances page, click a Kafka instance.
  2. В панели навигации выберите Instance > Topics.
  3. Нажмите Create Topic.
  4. Введите имя темы, укажите другие параметры, ссылаясь на Table 5, и нажмите OK.

    Table 5 Параметры темы

    Параметр

    Описание

    Имя темы

    Укажите имя, содержащее от 3 до 200 символов, начинающееся с буквы или символа подчёркивания (_), и содержащее только буквы, цифры, точки (.), дефисы (-) и подчёркивания (_).

    Имя должно отличаться от предустановленных тем:

    • __consumer_offsets
    • __transaction_state
    • __trace
    • __connect-status
    • __connect-configs
    • __connect-offsets
    • __cloud_event
    • __site_event
    • __dms_dial_test
    • __remote_log_metadata
    • __remote_log_compact_metadata
    • __remote_log_metadata_offset

    Cannot be changed once the topic is created.

    Введите topic-01.

    Разделы

    Если количество разделов совпадает с количеством потребителей, чем больше разделов, тем выше параллельность потребления.

    Введите 3.

    Реплики

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

    Введите 3.

    Время устаревания (ч)

    Как долго сообщения будут сохраняться в теме. Сообщения старше этого периода не могут быть потреблены. Они будут удалены, и их более нельзя будет потреблять.

    Введите 72.

    Synchronous Replication

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

    Synchronous Flushing

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

    Message Timestamp

    Выберите CreateTime: время, когда продюсер создал сообщение.

    Max. Message Size (bytes)

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

    Введите 10,485,760.

    Description

    Пропустить.

Шаг 4: Подключитесь к Kafka Instance для производства и потребления сообщений.

  1. Подготовьте файл для конфигурации производства и потребления.

    1. Войдите в Linux ECS.
    2. Скачайте сертификат client.jks и загрузите его в каталог /root на ECS.

      Чтобы получить сертификат: В консоли Kafka нажмите на экземпляр Kafka, чтобы перейти на страницу Basic Information. Нажмите Download рядом с SSL Certificate в области Connection. Распакуйте пакет, чтобы получить файл сертификата client.jks.
      Note

      /root — путь для хранения сертификата. При необходимости измените его на фактический путь.
    3. Перейдите в каталог /config на клиенте Kafka.
      cd kafka_2.12-2.7.2/config
    4. Добавьте следующие команды в оба файла consumer.properties и producer.properties (в качестве примера используется PLAIN).
      sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
      username="**********" \
      password="**********";        
      sasl.mechanism=PLAIN
      

      security.protocol=SASL_SSL
      ssl.truststore.location={ssl_truststore_path}
      ssl.truststore.password=dms@kafka
      ssl.endpoint.identification.algorithm=

      Описание:
      • username и password указываются при включении доступа к шифрованному тексту во время создания экземпляра.
      • ssl.truststore.location — путь для хранения сертификата, полученного в 1.b.
      • ssl.truststore.password сертифицируется сервером, его необходимо установить в dms@kafka и изменить нельзя.
      • ssl.endpoint.identification.algorithm определяет, проверять ли доменное имя сертификата. В данном примере оставьте этот параметр пустым, что означает отключение проверки доменного имени.

  2. Перейдите в каталог /bin на клиенте Kafka.

    cd ../bin

  3. Отправьте сообщения.

    ./kafka-console-producer.sh --broker-list {connection address} --topic {topic name} --producer.config ../config/producer.properties

    Описание:
    • {connection address}: адрес подключения, полученный в 10
    • {topic name}: имя темы, полученное в 4

    Например, 192.xxx.xxx.xxx:9093, 192.xxx.xxx.xxx:9093, 192.xxx.xxx.xxx:9093 являются адресами подключения Kafka instance.

    После выполнения этой команды вы можете отправлять сообщения в Kafka instance, вводя запрашиваемую информацию и нажимая Enter. Каждая строка содержимого будет отправлена как сообщение.

     
    [root@ecs-kafka bin]#./kafka-console-producer.sh --broker-list 192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093  --topic topic-01 --producer.config ../config/producer.properties
    >Hello
    >DMS
    >Kafka!
    >^C[root@ecs-kafka bin]#

    Нажмите Ctrl+C, чтобы отменить.

  4. Получать сообщения.

    ./kafka-console-consumer.sh --bootstrap-server {connection address} --topic {topic name} --from-beginning  --consumer.config ../config/consumer.properties

    Описание:
    • {connection address}: адрес подключения, полученный в 10
    • {topic name}: имя темы, полученное в 4

    Пример:

     
    [root@ecs-kafka bin]#  ./kafka-console-consumer.sh --bootstrap-server 192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093 --topic topic-01 --from-beginning --consumer.config ../config/consumer.properties
    Hello
    Kafka!
    DMS
    ^CProcessed a total of 3 messages
    [root@ecs-kafka bin]#

    Нажмите Ctrl+C, чтобы отменить.

Родительская тема: Getting Started
Предыдущая статья
Начало работы
Следующая статья
Процесс использования Kafka
Поддержка Юридические документы Политика конфиденциальности
© 2026 Cloud.ru