Работа с Kafka в TypeScript: KafkaJS примеры

Работа с Kafka в TypeScript с помощью KafkaJS

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

Что такое KafkaJS?

KafkaJS — это клиентская библиотека для Apache Kafka, написанная на JavaScript/TypeScript. Она поддерживает все основные операции: отправку и чтение сообщений, управление группами потребителей, работу с топиками и партициями. Благодаря встроенным типам, KafkaJS обеспечивает строгую типизацию и автодополнение в IDE, что значительно упрощает разработку и отладку.

Установка KafkaJS

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

npm install kafkajs

Если вы используете TypeScript, дополнительные типы не требуются — они уже включены в пакет.

Базовая настройка и подключение

Первый шаг — создание экземпляра клиента Kafka. Укажите идентификатор клиента и адреса брокеров:

import { Kafka } from 'kafkajs';

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

// Проверка подключения
const admin = kafka.admin();
await admin.connect();
console.log('Connected to Kafka');
await admin.disconnect();

Создание продюсера (Producer)

Продюсер отправляет сообщения в топики. KafkaJS поддерживает автоматическое создание топиков при отправке (если включено в конфигурации брокера).

import { Kafka, Partitioners } from 'kafkajs';

const producer = kafka.producer({
  createPartitioner: Partitioners.DefaultPartitioner
});

await producer.connect();

// Отправка одного сообщения
await producer.send({
  topic: 'orders',
  messages: [
    { key: 'order-123', value: JSON.stringify({ userId: 1, amount: 250 }) }
  ]
});

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

await producer.disconnect();

Создание консюмера (Consumer)

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

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

await consumer.connect();
await consumer.subscribe({ topic: 'orders', fromBeginning: true });

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

Работа с группами потребителей

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

// Два консюмера в одной группе будут читать разные партиции
const consumer1 = kafka.consumer({ groupId: 'order-group' });
const consumer2 = kafka.consumer({ groupId: 'order-group' });

// Каждый подписывается на один и тот же топик
await consumer1.subscribe({ topic: 'orders' });
await consumer2.subscribe({ topic: 'orders' });

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

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

Retry — автоматические повторные попытки при временных ошибках.
События — подписка на события consumer.connect, consumer.disconnect, consumer.crash.
Логирование — встроенный логгер с возможностью кастомизации.

import { logLevel } from 'kafkajs';

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

consumer.on('consumer.crash', async ({ error }) => {
  console.error('Consumer crashed:', error);
  // Логика восстановления
});