Kafka и Node.js: Event-Driven архитектура на практике

Введение

Event-Driven архитектура (EDA) позволяет компонентам системы взаимодействовать через события, не зная ничего друг о друге. Apache Kafka — один из наиболее популярных инструментов для реализации такого подхода в production-среде. В этой статье разберём, как интегрировать Kafka с Node.js: от настройки продюсера и консьюмера до обработки ошибок и управления offset-ами.

Что такое Event-Driven архитектура

В традиционной архитектуре сервисы общаются напрямую — один вызывает другой синхронно. EDA меняет этот подход: компоненты публикуют события, а другие компоненты подписываются на них. Это даёт три ключевых преимущества:

• Слабая связанность: сервисы не зависят друг от друга напрямую
• Масштабируемость: потребителей можно горизонтально масштабировать независимо
• Отказоустойчивость: сбой одного сервиса не блокирует остальные

Apache Kafka реализует эту модель через концепцию топиков (topics), партиций (partitions) и групп потребителей (consumer groups).

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

Начнём с локального запуска Kafka через Docker Compose:

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

Установим kafkajs — наиболее зрелый Kafka-клиент для Node.js:

npm install kafkajs

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

Продюсер публикует сообщения в топик. Создадим типизированный сервис:

import { Kafka, Producer } from 'kafkajs';

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

const producer: Producer = kafka.producer();

async function publishOrderEvent(orderId: string, status: string): Promise<void> {
  await producer.send({
    topic: 'order-events',
    messages: [
      {
        // ключ гарантирует порядок сообщений в рамках одной партиции
        key: orderId,
        value: JSON.stringify({
          orderId,
          status,
          timestamp: new Date().toISOString(),
        }),
      },
    ],
  });
}

await producer.connect();
await publishOrderEvent('order-123', 'CREATED');
await producer.disconnect();

Важный момент: key в сообщении определяет, в какую партицию попадёт запись. Все сообщения с одинаковым ключом гарантированно окажутся в одной партиции — это обеспечивает порядок обработки для конкретного объекта.

Создание консьюмера

Consumer Group позволяет нескольким экземплярам сервиса параллельно обрабатывать разные партиции:

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

const consumer: Consumer = kafka.consumer({
  // все экземпляры с одинаковым groupId образуют одну группу
  groupId: 'notification-group',
});

interface OrderEvent {
  orderId: string;
  status: string;
  timestamp: string;
}

async function processMessage({ message }: EachMessagePayload): Promise<void> {
  if (!message.value) return;

  const event: OrderEvent = JSON.parse(message.value.toString());
  console.log(`Получено событие: orderId=${event.orderId}, status=${event.status}`);

  // здесь размещается бизнес-логика
  await sendNotification(event);
}

await consumer.connect();
await consumer.subscribe({ topics: ['order-events'], fromBeginning: false });
await consumer.run({ eachMessage: processMessage });

Гарантии доставки и транзакции

Kafka поддерживает три режима гарантии доставки:

• at-most-once: сообщение может быть потеряно, но не дублировано
• at-least-once: сообщение доставится минимум один раз, возможны дубли
• exactly-once: ровно одна доставка, требует транзакций

Для критичных операций используйте транзакционного продюсера:

const txProducer = kafka.producer({
  transactionalId: 'order-tx-producer',
  maxInFlightRequests: 1, // обязательно для транзакций
  idempotent: true,       // предотвращает дубли при повторной отправке
});

await txProducer.connect();
const transaction = await txProducer.transaction();

try {
  await transaction.send({
    topic: 'order-events',
    messages: [{ key: 'order-456', value: JSON.stringify({ status: 'PAID' }) }],
  });
  await transaction.send({
    topic: 'payment-events',
    messages: [{ key: 'order-456', value: JSON.stringify({ amount: 1500 }) }],
  });
  // фиксируем обе записи атомарно
  await transaction.commit();
} catch (error) {
  await transaction.abort(); // откатываем при любой ошибке
  throw error;
}

Частые ошибки

1. Нет обработки ошибок в консьюмере

Если eachMessage выбрасывает необработанное исключение, KafkaJS останавливает всю группу потребителей. Оборачивайте логику в try/catch и явно решайте: повторить, отправить в dead-letter queue или пропустить сообщение.

2. Неверное число партиций

Число партиций определяет максимальный параллелизм: при 3 партициях максимум 3 консьюмера из одной группы работают одновременно. Уменьшить число партиций после создания топика нельзя — планируйте с запасом.

3. Игнорирование ручного commit offset

При autoCommit: true offset фиксируется до завершения обработки. При сбое сообщение будет потеряно. Для критичных данных используйте ручной commit:

await consumer.run({
  autoCommit: false,
  eachMessage: async (payload) => {
    await processMessage(payload);
    // фиксируем только после успешной обработки
    await consumer.commitOffsets([{
      topic: payload.topic,
      partition: payload.partition,
      offset: String(Number(payload.message.offset) + 1),
    }]);
  },
});

4. Утечки соединений

Всегда вызывайте disconnect() при завершении процесса:

process.on('SIGINT', async () => {
  await consumer.disconnect();
  process.exit(0);
});

Заключение

Event-Driven архитектура с Kafka открывает возможности для построения масштабируемых и отказоустойчивых Node.js-систем. Ключевые принципы: выбирайте гарантии доставки под конкретную задачу, планируйте партиции заранее, обрабатывайте ошибки явно и управляйте offset-ами осознанно. Начните с простого продюсера и консьюмера, а по мере роста нагрузки добавляйте транзакции и мониторинг через Kafka UI или Confluent Control Center.