Kafka em TypeScript: KafkaJS para Node.js

Biblioteca KafkaJS: trabalhando com Apache Kafka em TypeScript

\n\n

Apache Kafka é uma plataforma distribuída para processamento de fluxo de dados, amplamente utilizada para construir sistemas de troca de mensagens de alta carga. No ecossistema TypeScript (e Node.js), a biblioteca KafkaJS tornou-se o padrão de facto. Ela fornece uma API limpa e moderna com suporte total a TypeScript, permitindo que desenvolvedores escrevam código type-safe para produtores e consumidores Kafka.

\n\n

Neste artigo, abordaremos detalhadamente a instalação, configuração e os principais cenários de uso do KafkaJS. Você aprenderá como enviar e receber mensagens, tratar erros e configurar corretamente o cliente para ambientes de produção.

\n\n

Instalação e configuração do KafkaJS

\n\n

Para começar, instale a biblioteca via npm ou yarn:

\n\n

npm install kafkajs\n# ou\nyarn add kafkajs

\n\n

KafkaJS é escrito em TypeScript, portanto os tipos já estão embutidos — não é necessário instalar @types/kafkajs adicionalmente.

\n\n

Agora, vamos criar uma instância do cliente Kafka. Recomenda-se especificar vários brokers para tolerância a falhas:

\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, // Nível de log\n  retry: {\n    initialRetryTime: 100,\n    retries: 8\n  }\n});

\n\n

Produtor: enviando mensagens para o Kafka

\n\n

O produtor (producer) é responsável por escrever mensagens nos tópicos do Kafka. O KafkaJS fornece uma API conveniente para enviar mensagens individuais ou em lote.

\n\n

Exemplo de um produtor 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('Mensagem enviada');\n}\n\nasync function runProducer() {\n  const producer = kafka.producer();\n  await producer.connect();\n\n  // Envio de múltiplas mensagens\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: Sempre chame disconnect() após finalizar o trabalho para liberar recursos. Para aplicações de longa duração, o produtor pode ser mantido conectado permanentemente.

\n\n

Consumidor: lendo mensagens do Kafka

\n\n

O consumidor (consumer) se inscreve em um ou mais tópicos e processa as mensagens. O KafkaJS suporta grupos de consumidores para processamento paralelo.

\n\n

Exemplo de um consumidor com processamento de mensagens:

\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 // Ler desde a primeira mensagem\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

Observe o parâmetro fromBeginning: se definido como true, o consumidor lerá todas as mensagens acumuladas anteriormente. Para ambientes de produção, geralmente usa-se false (apenas mensagens novas).