Arbeit mit Apache Kafka in Node.js: Bibliothek KafkaJS

Arbeit mit Apache Kafka in Node.js mit der Bibliothek KafkaJS

\n\n

Apache Kafka ist eine verteilte Plattform für die Ereignisstromverarbeitung, die sich zum De-facto-Standard für den Aufbau hochbelastbarer Nachrichtenaustauschsysteme entwickelt hat. Im Node.js-Ökosystem ist KafkaJS eine der beliebtesten und stabilsten Bibliotheken für die Interaktion mit Kafka. Es handelt sich um einen modernen, vollständig asynchronen Client, der in reinem JavaScript geschrieben ist und alle wichtigen Kafka-Funktionen unterstützt: Producer, Consumer, Verbrauchergruppen, Offset-Verwaltung und Transaktionen.

\n\n

In diesem Artikel werden wir im Detail besprechen, wie man KafkaJS installiert, eine Verbindung zu einem Kafka-Cluster herstellt, Nachrichten sendet und liest, sowie die besten Praktiken für die Arbeit mit dieser Bibliothek in einer Produktionsumgebung betrachten.

\n\n

Installation und Konfiguration von KafkaJS

\n\n

Die Installation der Bibliothek erfolgt mit dem Standard-Paketmanager npm:

\n\n

npm install kafkajs

\n\n

Nach der Installation muss ein Client erstellt werden, der die Broker-Adressen und eine Client-ID angibt. Beispiel einer minimalen Konfiguration:

\n\n

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

\n\n

Für eine Produktionsumgebung wird empfohlen, mehrere Broker durch Kommas getrennt anzugeben, um Ausfallsicherheit zu gewährleisten, sowie SSL- und SASL-Authentifizierung zu konfigurieren:

\n\n

const kafka = new Kafka({\n  clientId: 'my-production-app',\n  brokers: ['broker1:9092', 'broker2:9092', 'broker3:9092'],\n  ssl: true,\n  sasl: {\n    mechanism: 'scram-sha-256',\n    username: 'myuser',\n    password: 'mypassword'\n  }\n});

\n\n

Producer: Senden von Nachrichten an ein Topic

\n\n

Der Producer ist für die Veröffentlichung von Nachrichten in einem angegebenen Kafka-Topic verantwortlich. KafkaJS bietet zwei Betriebsmodi für den Producer: einfach (send) und transaktional (für exakte Zustellung). Betrachten wir ein grundlegendes Beispiel zum Senden einer Nachricht:

\n\n

const producer = kafka.producer();\n\nasync function sendMessage() {\n  await producer.connect();\n\n  await producer.send({\n    topic: 'test-topic',\n    messages: [\n      { key: 'user1', value: 'Hello Kafka from Node.js!' },\n      { key: 'user2', value: JSON.stringify({ orderId: 123, amount: 2500 }) }\n    ]\n  });\n\n  await producer.disconnect();\n}\n\nsendMessage().catch(console.error);

\n\n

Wichtige Punkte:

\n

\n
• topic — der Name des Topics, an das die Nachricht gesendet wird.
\n
• messages — ein Array von Objekten, von denen jedes einen key (optional) und einen value (String oder Buffer) enthalten kann.
\n
• Es wird empfohlen, producer.connect() immer nur einmal beim Start der Anwendung aufzurufen und die Verbindung wiederzuverwenden.
\n

\n\n

Zur Leistungssteigerung in der Produktion verwenden Sie einen Verbindungspool und die acks-Konfiguration:

\n\n

const producer = kafka.producer({\n  allowAutoTopicCreation: false,\n  transactionTimeout: 30000\n});\n\nawait producer.send({\n  topic: 'orders',\n  messages: [{ value: 'New order' }],\n  acks: -1  // Warten auf Bestätigung von allen Replikaten\n});

\n\n

Consumer: Lesen von Nachrichten aus einem Topic

\n\n

Der Consumer ermöglicht das Lesen von Nachrichten aus einem oder mehreren Topics. Ein zentrales Konzept sind die Verbrauchergruppen (consumer groups), die die Lastverteilung sicherstellen und garantieren, dass jede Nachricht nur von einem Consumer aus der Gruppe verarbeitet wird.

\n\n

const consumer = kafka.consumer({ groupId: 'my-group' });\n\nasync function consumeMessages() {\n  await consumer.connect();\n  await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });\n\n  await consumer.run({\n    eachMessage: async ({ topic, partition, message }) => {\n      console.log({\n        partition,\n        offset: message.offset,\n        key: message.key.toString(),\n        value: message.valu