Консьюмеры¶

Консьюмер (consumer) — получатель сообщений из Kafka.

Справочник Консьюмеры имеет двухуровневую структуру:

Диспетчер задач — группа консьюмеров с общими параметрами параллелизма;
Консьюмер — настройки чтения для конкретного топика.

Диспетчер задач¶

Создаётся отдельной кнопкой в списке консьюмеров.

Поля диспетчера¶

Поле	Описание
Наименование	Произвольное название диспетчера
Подпотоки обработки	Количество параллельных потоков десериализации
Подпотоки транспорта	Количество параллельных потоков чтения из Kafka

Консьюмер¶

Произвольный обработчикКонвертация данныхДополнительно

Команды формы¶

Команда	Описание
Блокировка / Чтения	Останавливает получение новых сообщений из Kafka
Блокировка / Десериализации	Останавливает обработку полученных сообщений

Основные поля¶

Поле	Описание
Наименование	Произвольное название консьюмера
Брокер	Брокер Kafka, из которого читаются сообщения
Идентификатор	Идентификатор группы консьюмеров (`consumer group id`) — все консьюмеры с одним идентификатором получают разные сообщения из топика
Тайм-аут ожидания	Время ожидания сообщений от брокера в миллисекундах
Двоичные данные	Получать тело сообщения как двоичные данные (не преобразовывать в строку)
Идемпотентная обработка	Не обрабатывать повторно сообщения с одинаковым содержимым — см. ниже

Идемпотентная обработка¶

При включённом флаге адаптер перед десериализацией сравнивает SHA-256-хэш тела сообщения с хэшем последней успешной обработки по тому же ключу записи очереди (регистр «Хэш-суммы сообщений»). Если содержимое не изменилось — обработчик не вызывается: запись получает статус «Обработка отменена» с текстом «Сообщение уже обработано». Если содержимое изменилось — сообщение обрабатывается, а сохранённый хэш обновляется.

Настройка включается независимо для каждого консьюмера.
Сравнение выполняется только для одной и той же записи локальной очереди — текущее состояние против её предыдущей обработки.
Срок хранения истории хэшей (по умолчанию 3 дня) настраивается в поле Идемпотентность на вкладке Kafka / Администрирование / Очереди.

Подробнее о сценариях и способах форсировать повторную обработку — Идемпотентность.

Consumer group id

Если два консьюмера настроены с одним consumer group id и подписаны на один топик — Kafka будет распределять сообщения между ними (каждое сообщение уходит только одному). Для параллельной обработки одного потока сообщений несколькими консьюмерами используйте разные идентификаторы.

Топики¶

Из каких топиков читать и как обрабатывать:

Поле	Описание
Топик	Имя топика Kafka, из которого читаются сообщения
Тип десериализации	Способ обработки: Конвертация данных или Произвольный обработчик
Имя метода десериализации	Для «Произвольный обработчик» — имя экспортного метода
Имя модуля десериализации	Для «Произвольный обработчик» — имя общего модуля
Имя ПОД десериализации	Для «Конвертация данных» — имя правил обработки данных из модуля обмена КД 3.1
Формат десериализации	URL пространства имён XDTO-пакета

Автодополнение топиков

При открытии формы консьюмера список доступных топиков автоматически запрашивается из брокера Kafka через Admin API и становится доступен для выбора и автодополнения в поле Топик. При смене брокера список обновляется. Системные внутренние топики Kafka в список не включаются.

Параметры консьюмера¶

Дополнительные параметры librdkafka:

Поле	Описание
Ключ	Название параметра
Значение	Значение параметра