Kafka en TypeScript: KafkaJS para Node.js

Librería KafkaJS: trabajo con Apache Kafka en TypeScript

\n\n

Apache Kafka es una plataforma distribuida para el procesamiento de datos en streaming, ampliamente utilizada para construir sistemas de mensajería de alta carga. En el ecosistema de TypeScript (y Node.js), el estándar de facto es la librería KafkaJS. Proporciona una API limpia y moderna con soporte completo para TypeScript, lo que permite a los desarrolladores escribir código con seguridad de tipos para productores y consumidores de Kafka.

\n\n

En este artículo, analizaremos en detalle la instalación, configuración y los principales casos de uso de KafkaJS. Aprenderá cómo enviar y recibir mensajes, manejar errores y configurar correctamente el cliente para un entorno de producción.

\n\n

Instalación y configuración de KafkaJS

\n\n

Para empezar, instale la librería a través de npm o yarn:

\n\n

npm install kafkajs\n# o\nyarn add kafkajs

\n\n

KafkaJS está escrita en TypeScript, por lo que los tipos ya están integrados; no es necesario instalar @types/kafkajs adicionalmente.

\n\n

Ahora creemos una instancia del cliente Kafka. Se recomienda especificar varios brokers para alta disponibilidad:

\n\n

import { Kafka, logLevel } from 'kafkajs';\n\nconst kafka = new Kafka({\n  clientId: 'my-app',\n  brokers: ['localhost:9092', 'localhost:9093'],\n  logLevel: logLevel.INFO, // Nivel de registro\n  retry: {\n    initialRetryTime: 100,\n    retries: 8\n  }\n});

\n\n

Productor: envío de mensajes a Kafka

\n\n

El productor (producer) es responsable de escribir mensajes en los tópicos de Kafka. KafkaJS proporciona una API conveniente para enviar tanto mensajes individuales como lotes.

\n\n

Ejemplo de un productor básico:

\n\n

import { Kafka, Producer } from 'kafkajs';\n\nasync function produceMessage(producer: Producer) {\n  await producer.send({\n    topic: 'orders',\n    messages: [\n      { \n        key: 'order-123', \n        value: JSON.stringify({ userId: 42, amount: 1500 }) \n      },\n    ],\n  });\n  console.log('Mensaje enviado');\n}\n\nasync function runProducer() {\n  const producer = kafka.producer();\n  await producer.connect();\n\n  // Envío de varios mensajes\n  await producer.send({\n    topic: 'orders',\n    messages: [\n      { key: 'order-124', value: '{"item":"laptop"}' },\n      { key: 'order-125', value: '{"item":"mouse"}' },\n    ],\n  });\n\n  await produceMessage(producer);\n  await producer.disconnect();\n}\n\nrunProducer().catch(console.error);

\n\n

Importante: Siempre llame a disconnect() después de terminar para liberar recursos. Para aplicaciones de larga duración, el productor puede mantenerse conectado de forma permanente.

\n\n

Consumidor: lectura de mensajes desde Kafka

\n\n

El consumidor (consumer) se suscribe a uno o varios tópicos y procesa los mensajes. KafkaJS admite grupos de consumidores para procesamiento paralelo.

\n\n

Ejemplo de un consumidor con procesamiento de mensajes:

\n\n

import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';\n\nasync function runConsumer() {\n  const consumer: Consumer = kafka.consumer({ \n    groupId: 'order-processor-group' \n  });\n\n  await consumer.connect();\n  await consumer.subscribe({ \n    topic: 'orders', \n    fromBeginning: true // Leer desde el primer mensaje\n  });\n\n  await consumer.run({\n    eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {\n      const key = message.key?.toString();\n      const value = message.value?.toString();\n      console.log({\n        partition,\n        offset: message.offset,\n        key,\n        value: value ? JSON.parse(value) : null,\n      });\n    },\n  });\n}\n\nrunConsumer().catch(console.error);

\n\n

Tenga en cuenta el parámetro fromBeginning: si está configurado como true, el consumidor leerá todos los mensajes acumulados anteriormente. Para entornos de producción, normalmente se usa false (solo mensajes nuevos).