Arbeiten mit Kafka in TypeScript: Die Bibliothek KafkaJS für Node.js

Bibliothek KafkaJS: Arbeiten mit Apache Kafka in TypeScript

\n\n

Apache Kafka ist eine verteilte Plattform für die Stream-Verarbeitung von Daten, die häufig für den Aufbau hochbelastbarer Nachrichtensysteme verwendet wird. Im Ökosystem von TypeScript (und Node.js) hat sich die Bibliothek KafkaJS als De-facto-Standard etabliert. Sie bietet eine saubere, moderne API mit vollständiger TypeScript-Unterstützung, die es Entwicklern ermöglicht, typsicheren Code für Kafka-Producer und -Consumer zu schreiben.

\n\n

In diesem Artikel werden wir die Installation, Konfiguration und die wichtigsten Nutzungsszenarien von KafkaJS im Detail behandeln. Sie erfahren, wie Sie Nachrichten senden und empfangen, Fehler behandeln und den Client für die Produktionsumgebung richtig konfigurieren.

\n\n

Installation und Konfiguration von KafkaJS

\n\n

Installieren Sie die Bibliothek zunächst über npm oder yarn:

\n\n

npm install kafkajs\n# oder\nyarn add kafkajs

\n\n

KafkaJS ist in TypeScript geschrieben, daher sind die Typen bereits integriert – eine zusätzliche Installation von @types/kafkajs ist nicht erforderlich.

\n\n

Erstellen wir nun eine Instanz des Kafka-Clients. Es wird empfohlen, mehrere Broker für Ausfallsicherheit anzugeben:

\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, // Protokollierungsebene\n  retry: {\n    initialRetryTime: 100,\n    retries: 8\n  }\n});

\n\n

Producer: Senden von Nachrichten an Kafka

\n\n

Der Producer ist für das Schreiben von Nachrichten in Kafka-Themen zuständig. KafkaJS bietet eine praktische API zum Senden sowohl einzelner Nachrichten als auch von Batches.

\n\n

Beispiel eines einfachen Producers:

\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('Nachricht gesendet');\n}\n\nasync function runProducer() {\n  const producer = kafka.producer();\n  await producer.connect();\n\n  // Senden mehrerer Nachrichten\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

Wichtig: Rufen Sie nach Abschluss der Arbeit immer disconnect() auf, um Ressourcen freizugeben. Bei langlebigen Anwendungen kann der Producer dauerhaft verbunden bleiben.

\n\n

Consumer: Lesen von Nachrichten aus Kafka

\n\n

Der Consumer abonniert ein oder mehrere Themen und verarbeitet Nachrichten. KafkaJS unterstützt Consumergruppen für die parallele Verarbeitung.

\n\n

Beispiel eines Consumers mit Nachrichtenverarbeitung:

\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 // Von der ersten Nachricht an lesen\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

Beachten Sie den Parameter fromBeginning: Wenn er auf true gesetzt ist, liest der Consumer alle zuvor gespeicherten Nachrichten. In der Produktionsumgebung wird normalerweise false verwendet (nur neue Nachrichten).