Der Schlüssel zum Erstellen und Verbrauch von Apache -Kafka -Nachrichten mithilfe von Java liegt darin, die Hersteller- und Verbraucher -APIs ordnungsgem?? zu konfigurieren und ihre grundlegenden Prozesse zu verstehen. 1. Fügen Sie zuerst Kafka -Client -Abh?ngigkeiten hinzu, um sicherzustellen, dass die Version mit dem Cluster kompatibel ist. 2. Konfigurieren Sie beim Schreiben von Produzenten Bootstrap.Servers, Key.Serializer und Value.Serializer und erstellen Sie eine Kafkaproducer -Instanz, um Nachrichten zu senden, achten Sie auf das Schlie?en der Ressourcen und die optionale Rückrufverarbeitung. 3. Wenn Sie Verbraucher schreiben, Group.ID, Deserializer usw. konfigurieren, verwenden Sie Kafkaconsumer, um Themen zu abonnieren und Nachrichten in einer Schleife zu ziehen, auf die Strategie der übermittlung von Offsets achten. 4. H?ufige Probleme sind Gruppen.ID-Konflikte, unangemessene Einreichung von Offsets und ein falsches Ressourcenschlie? usw. Es wird empfohlen, den überprüfungsprozess zuerst in einer lokalen eigenst?ndigen Umgebung zu testen.

Es ist eigentlich nicht schwierig, Apache -Kafka -Nachrichten in Java zu produzieren und zu konsumieren. Der Kern liegt darin, die grundlegenden Konzepte von Kafka zu verstehen und die Verwendung von Produzenten- und Verbraucher -APIs zu beherrschen. Solange die Konfiguration ordnungsgem?? ist und die Codestruktur klar ist, kann das Senden und Empfangen von Nachrichten leicht erreicht werden.

Fügen Sie Kafka -Abh?ngigkeiten hinzu

Bevor Sie mit dem Schreiben von Code beginnen, müssen Sie die Kundenbibliothek von Kafka in Ihr Projekt einführen. Wenn Sie Maven verwenden, k?nnen Sie die folgenden Abh?ngigkeiten in pom.xml hinzufügen:

 <De vorangehen>
    <gruppe> org.apache.kafka </Groupid>
    <artifactid> Kafka-Clients </artifactid>
    <version> 3.6.0 </Version>
</abh?ngig>

Diese Version ist relativ neu und relativ stabil. Natürlich k?nnen Sie auch die entsprechende Client -Version basierend auf Ihrer Kafka -Clusterversion ausw?hlen.

Schreiben Sie einen Kafka -Produzenten

Die Aufgabe des Produzenten besteht darin, Nachrichten an das Kafka -Thema zu senden. Zu den Hauptschritten geh?rt das Konfigurieren von Eigenschaften, das Erstellen einer Produzenteninstanz, das Erstellen eines Nachrichtendatensatzes und das Senden.

Schlüsselkonfigurationselemente sind:

• bootstrap.servers : Kafka Broker -Adresse
• key.serializer und value.serializer : Geben Sie die Serialisierungsmethode von Schlüssel und Wert an. StringSerializer wird im Allgemeinen verwendet.

Der Beispielcode lautet wie folgt:

 Eigenschaften props = neue Eigenschaften ();
props.put ("bootstrap.servers", "localhost: 9092");
props.put ("key.serializer", "org.apache.kafka.common.serialization.Stringserializer");
props.put ("value.serializer", "org.apache.kafka.common.Serialization.Stringserializer");

Produzent <String, String> Produzent = neuer KafkaproDucer <> (Requisiten);
ProducTeRecord <String, String> record = new ProducerRecord <> ("my-topic", "key", "value");

Produzent.Send (Record);
Produzent.CLOSE ();

Einige Punkte zu beachten:

• Nach dem Senden ist es am besten, close() anzurufen, um Ressourcenleckage zu vermeiden
• Wenn Sie best?tigen m?chten, ob das Senden erfolgreich ist, k?nnen Sie die Callback -Funktion addieren .send(record, callback)
• Wenn es sich nur um einen Test handelt, k?nnen Sie den Schlüssel weglassen

Schreiben Sie den Kafka -Verbraucher

Der Verbraucher ist dafür verantwortlich, Nachrichten aus dem Thema Kafka zu lesen. Im Vergleich zu Produzenten ist die Logik des Verbrauchers etwas komplizierter, da sie aktiv an Nachrichten, Offsets usw. verarbeitet werden müssen.

Zu den wichtigsten Konfigurationen geh?ren:

• bootstrap.servers : Geben Sie auch die Broker -Adresse an
• group.id : Verbrauchergruppen -ID, muss festgelegt werden, andernfalls wird ein Fehler gemeldet
• key.deserializer und value.deserializer : Deserializer, normalerweise mit StringDeserializer

Ein einfacher Verbraucherprozess ist wie folgt:

 Eigenschaften props = neue Eigenschaften ();
props.put ("bootstrap.servers", "localhost: 9092");
props.put ("Gruppe.id", "Testgruppe");
props.put ("key.deserializer", "org.apache.kafka.common.serialization.stringDeserializer");
props.put ("value.deserializer", "org.apache.kafka.common.serialization.stringDeserializer");

Kafkaconsumer <String, String> Consumer = new Kafkaconsumer <> (Requisiten);
Consumer.

while (wahr) {
    CENSPEERRECORDS <String, String> Records = Consumer.Poll (Duration.ofmillis (100));
    für (CENSELERRECORD <STRING, STRING> DRECKREISE: Datens?tze) {
        System.out.println ("empfangen:" record.Value ());
    }
}

Notiz:

• poll() -Methode zieht eine Menge Daten und verarbeitet sie in einer Schleife
• Vergessen Sie nicht, nach dem Verbrauch einen Offset einzureichen. Sie k?nnen w?hlen, ob Sie automatisch oder manuell einreichen m?chten
• Wenn Sie nur einmal konsumieren und dann beenden m?chten, k?nnen Sie die Schleife nach der Verarbeitung brechen

H?ufig gestellte Fragen und Vorsichtsma?nahmen

Manchmal werden Sie feststellen, dass die Nachricht nicht wiederholt empfangen oder konsumiert wurde, was normalerweise aus den folgenden Gründen verursacht wird:

• Consumer Group.ID Einstellungsfehler oder Konflikt
• Die automatische übermittlung der Versatzfrequenz ist zu hoch oder zu niedrig
• Das Vers?umnis, den Hersteller oder Verbraucher nicht zu schlie?en, führt zu einem inkonsistenten Status
• Der Kafka -Broker wird nicht gestartet oder das Netzwerk ist nicht verbunden

Es wird empfohlen, zuerst die lokale eigenst?ndige Umgebung w?hrend der Entwicklung zu leiten und dann zum Cluster zu gehen.

Grunds?tzlich ist das. Der Prozess der Verbindung zu Kafka in Java ist nicht kompliziert, aber einige Details sind leicht zu ignorieren, wie z. B. Serialisierung, Gruppe.ID und Einreichungsstrategien. Solange Sie auf die Konfiguration und den Prozess achten, k?nnen Sie schnell ausführen.

Das obige ist der detaillierte Inhalt vonWie produziere und konsumiere ich Nachrichten von Apache Kafka mit Java?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!