Kafka в Node.js (JavaScript): KafkaJS | Руководство |…

Работа с Apache Kafka в Node.js с помощью библиотеки KafkaJS

Apache Kafka — это распределённая платформа для потоковой передачи событий, ставшая стандартом де-факто для построения высоконагруженных систем обмена сообщениями. В экосистеме Node.js одной из самых популярных и стабильных библиотек для взаимодействия с Kafka является KafkaJS. Это современный, полностью асинхронный клиент, написанный на чистом JavaScript, который поддерживает все ключевые возможности Kafka: продюсеров, консьюмеров, группы потребителей, управление смещениями (offsets) и транзакции.

В этой статье мы подробно разберём, как установить KafkaJS, настроить подключение к кластеру Kafka, отправить и прочитать сообщения, а также рассмотрим лучшие практики работы с этой библиотекой в production-среде.

Установка и настройка KafkaJS

Установка библиотеки выполняется стандартным менеджером пакетов npm:

npm install kafkajs

После установки необходимо создать клиент, указав адреса брокеров и идентификатор клиента. Пример минимальной конфигурации:

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

Для production-среды рекомендуется указывать несколько брокеров через запятую для обеспечения отказоустойчивости, а также настраивать SSL и SASL-аутентификацию:

const kafka = new Kafka({
  clientId: 'my-production-app',
  brokers: ['broker1:9092', 'broker2:9092', 'broker3:9092'],
  ssl: true,
  sasl: {
    mechanism: 'scram-sha-256',
    username: 'myuser',
    password: 'mypassword'
  }
});

Продюсер: отправка сообщений в топик

Продюсер отвечает за публикацию сообщений в указанный топик Kafka. KafkaJS предоставляет два режима работы продюсера: простой (send) и транзакционный (для точной доставки). Рассмотрим базовый пример отправки сообщения:

const producer = kafka.producer();

async function sendMessage() {
  await producer.connect();

  await producer.send({
    topic: 'test-topic',
    messages: [
      { key: 'user1', value: 'Hello Kafka from Node.js!' },
      { key: 'user2', value: JSON.stringify({ orderId: 123, amount: 2500 }) }
    ]
  });

  await producer.disconnect();
}

sendMessage().catch(console.error);

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

topic — имя топика, в который отправляется сообщение.
messages — массив объектов, каждый из которых может содержать key (опционально) и value (строка или Buffer).
Рекомендуется всегда вызывать producer.connect() один раз при старте приложения и переиспользовать подключение.

Для повышения производительности в production используйте пул соединений и настройку acks:

const producer = kafka.producer({
  allowAutoTopicCreation: false,
  transactionTimeout: 30000
});

await producer.send({
  topic: 'orders',
  messages: [{ value: 'New order' }],
  acks: -1  // Ожидание подтверждения от всех реплик
});

Консьюмер: чтение сообщений из топика

Консьюмер позволяет читать сообщения из одного или нескольких топиков. Важнейшая концепция — группы потребителей (consumer groups), которые обеспечивают балансировку нагрузки и гарантируют, что каждое сообщение будет обработано только одним консьюмером из группы.

const consumer = kafka.consumer({ groupId: 'my-group' });

async function consumeMessages() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        partition,
        offset: message.offset,
        key: message.key.toString(),
        value: message.value.toString()
      });
    }
  });
}

consumeMessages().catch(console.error);

Пояснения к коду:

groupId — идентификатор группы потребителей. Консьюмеры с одинаковым groupId распределяют партиции между собой.
fromBeginning — если true, чтение начнётся с самого первого сообщения в топике (если смещение не сохранено).
Метод eachMessage вызывается для каждого полученного сообщения. Внутри него можно выполнять бизнес-логику, сохранять данные в БД и т.д.

Обработка ошибок и повторные попытки

В реальных приложениях важно корректно обрабатывать ошибки подключения, таймауты и сбои при обработке сообщений. KafkaJS предоставляет встроенные механизмы повторных попыток