Работа с Kafka в JavaScript (Node.js): библиотека node-rdkafka

Apache Kafka и Node.js: знакомство с node-rdkafka

Apache Kafka — это распределённая платформа для стриминга событий, ставшая стандартом де-факто для построения надёжных и масштабируемых систем обмена данными. В экосистеме Node.js одной из самых производительных и зрелых библиотек для работы с Kafka является node-rdkafka. Она представляет собой нативные биндинги к библиотеке librdkafka (написанной на C/C++), что обеспечивает максимальную скорость и стабильность.

В этой статье мы подробно разберём, как настроить и использовать node-rdkafka для отправки (producer) и чтения (consumer) сообщений. Вы узнаете, как интегрировать Kafka в ваше Node.js-приложение, обрабатывать ошибки и управлять потоками данных.

Установка node-rdkafka

Для установки библиотеки требуется Node.js версии 12 или выше и компилятор C++ (gcc, clang или MSVC). Установка производится стандартным менеджером пакетов:

npm install node-rdkafka

Если сборка завершается с ошибкой, убедитесь, что у вас установлены build-essential (Linux) или Python и инструменты сборки для Windows (npm install --global windows-build-tools).

Создание Producer (отправителя сообщений)

Producer в Kafka отвечает за публикацию сообщений в топики. Рассмотрим базовый пример:

const Kafka = require('node-rdkafka');

// Создаём продюсера
const producer = new Kafka.Producer({
  'metadata.broker.list': 'localhost:9092',
  'dr_cb': true  // Callback для подтверждения доставки
});

// Обработчик событий при подключении
producer.on('event.log', function(event) {
  console.log('Лог:', event.message);
});

// Обработчик ошибок
producer.on('event.error', function(err) {
  console.error('Ошибка продюсера:', err);
});

// Подключаемся к брокеру
producer.connect();

// После подключения отправляем сообщения
producer.on('ready', function() {
  try {
    // Отправка сообщения в топик 'test-topic'
    producer.produce(
      'test-topic',  // имя топика
      null,          // партиция (null = автоматическое распределение)
      Buffer.from('Привет, Kafka!'),  // тело сообщения (Buffer)
      'my-key',      // ключ сообщения (опционально)
      Date.now()     // timestamp (опционально)
    );
    console.log('Сообщение отправлено');
  } catch (err) {
    console.error('Ошибка при отправке:', err);
  }
});

// Не забываем вызывать poll для обработки callback'ов
setInterval(() => {
  producer.poll();
}, 1000);

// Закрытие соединения при завершении процесса
process.on('SIGINT', () => {
  producer.disconnect();
  process.exit();
});

Обратите внимание: метод producer.produce() принимает тело сообщения как Buffer. Это позволяет передавать как строки, так и бинарные данные. Параметр dr_cb: true включает callback-уведомления о доставке (delivery report).

Создание Consumer (получателя сообщений)

Consumer подписывается на один или несколько топиков и обрабатывает входящие сообщения. Для работы с группами потребителей используется KafkaConsumer:

const Kafka = require('node-rdkafka');

const consumer = new Kafka.KafkaConsumer({
  'group.id': 'my-group',
  'metadata.broker.list': 'localhost:9092',
  'enable.auto.commit': true
}, {});

consumer.on('event.error', function(err) {
  console.error('Ошибка консьюмера:', err);
});

consumer.on('ready', function() {
  console.log('Консьюмер готов');
  // Подписываемся на топик
  consumer.subscribe(['test-topic']);
  // Запускаем цикл потребления
  consumer.consume();
});

// Обработка каждого сообщения
consumer.on('data', function(data) {
  console.log('Получено сообщение:');
  console.log('  Топик:', data.topic);
  console.log('  Партиция:', data.partition);
  console.log('  Ключ:', data.key ? data.key.toString() : null);
  console.log('  Значение:', data.value.toString());
  console.log('  Timestamp:', data.timestamp);
});

consumer.connect();

// Закрытие
process.on('SIGINT', () => {
  consumer.disconnect();
  process.exit();
});

Ключевые моменты:

• group.id — идентификатор группы потребителей. Все консьюмеры с одинаковым group.id распределяют нагрузку между собой.
• enable.auto.commit — автоматическое сохранение смещений (offset). Для ручного управления используйте consumer.commit().
• Событие data срабатывает для каждого полученного сообщения.

Работа с партициями и управление offset

Kafka позволяет гибко управлять чтением сообщений. Вы можете указать конкретные партиции и смещения:

// Назначить конкретные партиции для чтения
consumer.assign([
  { topic