Arbeiten mit Kafka in TypeScript: KafkaJS Beispiele |…

Arbeiten mit Kafka in TypeScript mit KafkaJS

Apache Kafka ist eine verteilte Plattform für die Datenstromverarbeitung, die häufig in der Microservice-Architektur, der Ereignisverarbeitung und der Echtzeitanalyse eingesetzt wird. Bei der Entwicklung mit TypeScript ist KafkaJS eine der beliebtesten Bibliotheken für die Interaktion mit Kafka. Sie bietet eine moderne, asynchrone API mit vollständiger TypeScript-Unterstützung, was sie zur idealen Wahl für die Erstellung zuverlässiger und skalierbarer Anwendungen macht.

Was ist KafkaJS?

KafkaJS ist eine Client-Bibliothek für Apache Kafka, geschrieben in JavaScript/TypeScript. Sie unterstützt alle grundlegenden Operationen: Senden und Lesen von Nachrichten, Verwaltung von Verbrauchergruppen, Arbeit mit Topics und Partitionen. Dank integrierter Typen bietet KafkaJS eine strenge Typisierung und Autovervollständigung in der IDE, was die Entwicklung und Fehlersuche erheblich vereinfacht.

Installation von KafkaJS

Um zu beginnen, installieren Sie die Bibliothek über npm:

npm install kafkajs

Wenn Sie TypeScript verwenden, sind keine zusätzlichen Typen erforderlich – sie sind bereits im Paket enthalten.

Grundlegende Konfiguration und Verbindung

Der erste Schritt ist die Erstellung einer Instanz des Kafka-Clients. Geben Sie die Client-ID und die Adressen der Broker an:

import { Kafka } from 'kafkajs';

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

// Verbindung prüfenconst admin = kafka.admin();await admin.connect();console.log('Connected to Kafka');await admin.disconnect();

Erstellen eines Producers

Der Producer sendet Nachrichten an Topics. KafkaJS unterstützt die automatische Erstellung von Topics beim Senden (falls in der Broker-Konfiguration aktiviert).

import { Kafka, Partitioners } from 'kafkajs';

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

await producer.connect();

// Senden einer einzelnen Nachrichtawait producer.send({  topic: 'orders',  messages: [    { key: 'order-123', value: JSON.stringify({ userId: 1, amount: 250 }) }  ]});

// Senden mehrerer Nachrichtenawait producer.send({  topic: 'orders',  messages: [    { key: 'order-124', value: 'Message 2' },    { key: 'order-125', value: 'Message 3' }  ]});

await producer.disconnect();

Erstellen eines Consumers

Der Consumer liest Nachrichten aus Topics. Er muss auf ein oder mehrere Topics abonniert und in einer Verbrauchergruppe (consumer group) registriert sein.

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(),    });  },});

Arbeiten mit Verbrauchergruppen

Verbrauchergruppen ermöglichen die Lastverteilung zwischen mehreren Consumer-Instanzen. Kafka weist die Partitionen eines Topics automatisch verschiedenen Consumern innerhalb derselben Gruppe zu. Fällt ein Consumer aus, werden seine Partitionen neu verteilt.

// Zwei Consumer in derselben Gruppe lesen unterschiedliche Partitionenconst consumer1 = kafka.consumer({ groupId: 'order-group' });const consumer2 = kafka.consumer({ groupId: 'order-group' });

// Jeder abonniert dasselbe Topicawait consumer1.subscribe({ topic: 'orders' });await consumer2.subscribe({ topic: 'orders' });