Arbeit mit Kafka in Java: Bibliotheken, Konfiguration und Codebeispiele

Arbeit mit Kafka in Java: Bibliotheken, Konfiguration und Codebeispiele

Apache Kafka ist eine verteilte Plattform für die Datenstromverarbeitung, die häufig für den Aufbau hochbelastbarer Nachrichtensysteme, das Sammeln von Logs, die Echtzeit-Ereignisverarbeitung und die Integration von Microservices eingesetzt wird. Im Java-Ökosystem nimmt Kafka aufgrund der offiziellen Bibliothek Kafka Clients und vieler zusätzlicher Tools eine besondere Stellung ein. In diesem Artikel werden wir detailliert erläutern, wie man mit Kafka in Java arbeitet, welche Bibliotheken verwendet werden, und praktische Codebeispiele geben.

Wichtige Bibliotheken für die Arbeit mit Kafka in Java

Für die Integration von Java-Anwendungen mit Apache Kafka gibt es mehrere wichtige Bibliotheken:

\n
• Kafka Clients — die offizielle Bibliothek von Apache, die eine API für Producer und Consumer bereitstellt. Dies ist die minimalste und zuverlässigste Methode zur Interaktion mit Brokern.
\n
• Spring Kafka — eine Erweiterung für das Spring Framework, die die Konfiguration und Nutzung von Kafka durch Annotationen und Vorlagen (KafkaTemplate, @KafkaListener) vereinfacht.
\n
• Kafka Streams — eine Bibliothek zum Erstellen von Streaming-Anwendungen (Stream Processing) in Java, ohne dass Spark oder Flink erforderlich sind. Sie ermöglicht das Filtern, Aggregieren und Zusammenführen von Streams.
\n
• Akka Streams Kafka (Alpakka) — eine reaktive Bibliothek für die Arbeit mit Kafka auf Basis von Akka Streams, geeignet für asynchrone und nicht-blockierende Anwendungen.
\n

In diesem Leitfaden konzentrieren wir uns auf die offizielle Bibliothek Kafka Clients, da sie die Grundlage für alle anderen Tools bildet.

Abhängigkeiten einbinden (Maven)

Um zu beginnen, fügen Sie die folgende Abhängigkeit zu Ihrer pom.xml hinzu:

<dependency>\n    <groupId>org.apache.kafka</groupId>\n    <artifactId>kafka-clients</artifactId>\n    <version>3.7.0</version>\n</dependency>

Wenn Sie Gradle verwenden, fügen Sie hinzu:

implementation 'org.apache.kafka:kafka-clients:3.7.0'

Konfiguration des Producers

Der Producer sendet Nachrichten an Kafka-Topics. Betrachten wir ein einfaches Beispiel zum Erstellen und Senden einer Nachricht.

import org.apache.kafka.clients.producer.*;\nimport java.util.Properties;\n\npublic class SimpleProducer {\n    public static void main(String[] args) {\n        // Konfiguration der Producer-Eigenschaften\n        Properties props = new Properties();\n        props.put("bootstrap.servers", "localhost:9092");\n        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");\n        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");\n        props.put("acks", "all"); // Warten auf Bestätigung von allen Replikaten\n        props.put("retries", 3);  // Anzahl der Wiederholungsversuche bei Fehlern\n\n        // Erstellen des Producers\n        Producer<String, String> producer = new KafkaProducer<>(props);\n\n        // Senden einer Nachricht\n        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hallo, Kafka von Java!");\n        producer.send(record, (metadata, exception) -> {\n            if (exception == null) {\n                System.out.println("Nachricht gesendet an Topic: " + metadata.topic() +\n                                   ", Partition: " + metadata.partition() +\n                                   ", Offset: " + metadata.offset());\n            } else {\n                System.err.println("Fehler beim Senden: " + exception.getMessage());\n            }\n        });\n\n        // Schließen des Producers\n        producer.close();\n    }\n}

In diesem Beispiel geben wir die Broker-Adresse (localhost:9092), Serialisierer für Schlüssel und Wert (in die