Работа с Kafka в TypeScript: библиотека KafkaJS для Node.js

Библиотека KafkaJS: работа с Apache Kafka в TypeScript

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

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

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

Для начала работы установите библиотеку через npm или yarn:

npm install kafkajs
# или
yarn add kafkajs

KafkaJS написана на TypeScript, поэтому типы уже встроены — дополнительно устанавливать @types/kafkajs не требуется.

Теперь создадим экземпляр клиента Kafka. Рекомендуется указывать несколько брокеров для отказоустойчивости:

import { Kafka, logLevel } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092', 'localhost:9093'],
  logLevel: logLevel.INFO, // Уровень логирования
  retry: {
    initialRetryTime: 100,
    retries: 8
  }
});

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

Продюсер (producer) отвечает за запись сообщений в топики Kafka. KafkaJS предоставляет удобный API для отправки как отдельных сообщений, так и батчей.

Пример базового продюсера:

import { Kafka, Producer } from 'kafkajs';

async function produceMessage(producer: Producer) {
  await producer.send({
    topic: 'orders',
    messages: [
      { 
        key: 'order-123', 
        value: JSON.stringify({ userId: 42, amount: 1500 }) 
      },
    ],
  });
  console.log('Сообщение отправлено');
}

async function runProducer() {
  const producer = kafka.producer();
  await producer.connect();

  // Отправка нескольких сообщений
  await producer.send({
    topic: 'orders',
    messages: [
      { key: 'order-124', value: '{"item":"laptop"}' },
      { key: 'order-125', value: '{"item":"mouse"}' },
    ],
  });

  await produceMessage(producer);
  await producer.disconnect();
}

runProducer().catch(console.error);

Важно: Всегда вызывайте disconnect() после завершения работы, чтобы освободить ресурсы. Для долгоживущих приложений продюсер можно держать подключённым постоянно.

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

Консьюмер (consumer) подписывается на один или несколько топиков и обрабатывает сообщения. KafkaJS поддерживает группы консьюмеров для параллельной обработки.

Пример консьюмера с обработкой сообщений:

import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';

async function runConsumer() {
  const consumer: Consumer = kafka.consumer({ 
    groupId: 'order-processor-group' 
  });

  await consumer.connect();
  await consumer.subscribe({ 
    topic: 'orders', 
    fromBeginning: true // Читать с самого первого сообщения
  });

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

runConsumer().catch(console.error);

Обратите внимание на параметр fromBeginning: если установлен true, консьюмер прочитает все ранее накопленные сообщения. Для продакшн-среды обычно используют false (только новые сообщения).

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

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

Пример обработки ошибок консьюмера:

consumer.on('consumer.crash', async (error) => {
  console.error('Консьюмер упал:', error);
  // Можно переподключиться
  await consumer.connect();
});

consumer.on('consumer.disconnect', () => {
  console.warn('Консьюмер отключён');
});

Для продюсера можно настроить политику повторных попыток при создании клиента (параметр retry), а также обрабатывать ошибки отправки:

try {
  await producer.send({ topic: 'orders', messages: [...] });
} catch (error) {
  console.error('Ошибка отправки:', error);