Konfigurationseinstellungen

Für den Producer gibt es viele Konfigurationseinstellungen. Sie können Aspekte des Produzenten steuern, einschließlich Stapelverarbeitung, Wiederholungsversuche und Nachrichtenbestätigung. Dies hier sind die wichtigsten:

Es gibt noch viele weitere Konfigurationseinstellungen, aber lesen Sie die Apache Kafka gründlich durch, bevor Sie damit experimentieren.

Partitionierung

Bei Kafka stellen Partitionen die Einheiten der Skalierbarkeit dar. Daher ist die Partitionierung ein effektiver Weg, um den Durchsatz zu erhöhen, da sie es ermöglicht, dass die thematischen Daten in mehreren parallelen Strömen fließen.

Wenn der Producer eine Nachricht zu einem Topic veröffentlicht, wählt der Producer die zu verwendende Partition aus. Wenn Ordnung wichtig ist, beachten Sie, dass eine Partition eine geordnete Folge von Datensätzen ist, ein Thema jedoch eine oder mehrere Partitionen umfasst. Wenn eine Reihe von Nachrichten in der richtigen Reihenfolge zugestellt werden soll, stellen Sie sicher, dass sie alle zur gleichen Partition gehören. Die einfachste Möglichkeit, um dies zu erreichen, ist es, wenn alle Nachrichten denselben Schlüssel haben.

Der Producer kann explizit eine Partitionsnummer angeben, wenn er eine Nachricht veröffentlicht. Somit haben Sie eine direkte Kontrolle. Aber der Producercode wird dadurch auch komplexer, da er für die Partitionsauswahl verwaltet. Weitere Informationen dazu finden im Methodenaufruf "Producer.partitionsFor". Zum Beispiel wird der Aufruf beschrieben für Kafka Version 2.2.0.

Wenn der Producer keine Partitionsnummer angibt, wird die Auswahl der Partition durch eine Partitionierungsfunktion getroffen. Die Standard-Partitionierungsfunktion, die im Kafka-Producer enthalten ist, funktioniert wie folgt:

• Wenn der Datensatz keinen Schlüssel enthält, wählen Sie die Partition im Round-Robin-Modus aus.
• Wenn der Datensatz einen Schlüssel enthält, wählen Sie die Partition aus, indem Sie einen Hashwert für den Schlüssel berechnen. Dadurch wird für alle Nachrichten mit demselben Schlüssel dieselbe Partition ausgewählt.

Sie können auch Ihre eigene angepasste Partitionierungsfunktion schreiben. Eine angepasste Partitionierungsfunktion kann ein beliebiges Schema für die Zuordnung von Datensätzen zu Partitionen auswählen. Verwenden Sie zum Beispiel nur ein Subset der Informationen im Schlüssel oder eine anwendungsspezifische ID.

Nachrichtenreihenfolge

Kafka schreibt die Nachrichten in der Regel in der Reihenfolge, in der sie vom Producer gesendet werden. In bestimmten Situationen können Wiederholungsversuche jedoch dazu führen, dass Nachrichten dupliziert oder neu sortiert werden. Wenn Sie möchten, dass eine Folge von Nachrichten in der richtigen Reihenfolge gesendet wird, müssen Sie sicherstellen, dass sie alle in dieselbe Partition geschrieben werden, da nur so die Reihenfolge der Nachrichten gewährleistet werden kann.

Der Produzent kann auch versuchen, die Nachrichten automatisch zu senden. Es ist eine gute Idee, diese Wiederholungsfunktion zu aktivieren, denn die Alternative ist, dass Ihr Anwendungscode alle Wiederholungsversuche selbst durchführen muss. Die Kombination aus Stapelverarbeitung in Kafka und automatischen Neuversuchen kann dazu führen, dass Nachrichten dupliziert werden und deren Reihenfolge geändert wird.

Beispiel: Sie veröffentlichen eine Folge von drei Nachrichten <M1, M2, M3> zu einem Thema. Die Datensätze passen möglicherweise alle in denselben Stapel, so dass sie alle zusammen an den Partitionsleiter gesendet werden. Der Leader schreibt die Daten dann zur Partition und repliziert sie als separate Datensätze. Wenn ein Fehler auftritt, ist es möglich, dass M1 und M2 zur Partition hinzugefügt werden, M3 jedoch nicht. Der Erzeuger erhält keine Bestätigung und versucht daher erneut, <M1, M2, M3> zu senden. Der neue Leader schreibt M1, M2 und M3 in die Partition, die nun <M1, M2, M1, M2, M3> enthält, wobei das duplizierte M1 auf das ursprüngliche M2 folgt. Wenn Sie die Anzahl der gerade ausgeführten Anforderungen pro Broker auf einen Broker beschränken, können Sie die Änderung der Reihenfolge verhindern. Möglicherweise stellen Sie immer noch fest, dass ein einzelner Datensatz dupliziert wird, z. B. <M1, M2, M2, M3>, aber Sie erhalten nie Reihenfolgen. In Kafka ab Version 0.11 können Sie auch die Funktion des idempotenten Produzenten verwenden, um die Duplizierung von M2 zu verhindern.

Bei Kafka ist es üblich, die Anwendungen so zu schreiben, dass sie mit gelegentlichen Nachrichtenduplikaten umgehen können, da die Auswirkungen auf die Leistung, wenn nur eine einzige Anfrage im Umlauf ist, erheblich sind.

Bestätigung von Nachrichten

Wenn Sie eine Nachricht veröffentlichen, können Sie mit Hilfe der acks Producer-Konfiguration festlegen, wie viele Bestätigungen erforderlich sind. Diese Auswahl ist ein Ausgleich zwischen Durchsatz und Zuverlässigkeit. Die folgenden drei Ebenen sind vorhanden.

acks=0 (am wenigsten zuverlässig)

Die Nachricht gilt als gesendet, sobald sie in das Netz geschrieben wurde. Es gibt keine Bestätigung vom Partitionsleader. Daher können Nachrichten verloren gehen, wenn sich das Leadership der Partition ändert. Diese Art der Bestätigung ist schnell, birgt aber in manchen Situationen die Möglichkeit eines Nachrichtenverlusts.

acks=1 (Standardoption)

Die Nachricht wird dem Produzenten bestätigt, sobald der Partition Leader seinen Datensatz erfolgreich in die Partition geschrieben hat. Da die Bestätigung erfolgt, bevor der Datensatz die synchronisierten Replikate erreicht hat, kann die Nachricht verloren gehen, wenn der Leader ausfällt, die Follower die Nachricht aber noch nicht haben. Ändert sich die Führung der Partition, informiert der alte Leader den Producer, der den Fehler beheben und erneut versuchen kann, die Nachricht an den neuen Leader zu senden. Da Nachrichten bestätigt werden, bevor ihr Empfang von allen Replikaten bestätigt wurde, können Nachrichten, die zwar bestätigt, aber noch nicht vollständig repliziert wurden, verloren gehen, wenn sich die Partitionsführung ändert.

acks=all (am zuverlässigsten)

Die Nachricht wird dem Produzenten bestätigt, wenn der Partition Leader seinen Datensatz erfolgreich geschrieben hat und alle in-synchronen Replikate dasselbe getan haben. Die Nachricht geht nicht verloren, wenn sich das Leadership der Partition ändert. Voraussetzung ist aber, dass mindestens ein synchrones Replikat verfügbar ist.

Auch wenn Sie nicht darauf warten, dass Nachrichten an den Producer bestätigt werden, sind die Nachrichten nur dann zur Verarbeitung verfügbar, wenn sie per Commit festgeschrieben werden. Das bedeutet, dass die Replikation mit den synchronen Replikaten abgeschlossen ist. Mit anderen Worten: Die Latenzzeit für das Senden der Nachrichten aus Sicht des Producers ist geringer als die Endpunkt-zu-Endpunkt-Latenzzeit, die gemessen wird, wenn der Producer eine Nachricht an einen Consumer sendet, der diese Nachricht empfängt.

Warten Sie nach Möglichkeit nicht auf die Bestätigung einer Nachricht, bevor Sie die nächste Nachricht veröffentlichen. Durch das Warten wird verhindert, dass der Producer Nachrichten im Stapel zusammenfassen kann. Außerdem wird die Rate reduziert, mit der Nachrichten unterhalb der Round-Trip-Latenz des Netzes veröffentlicht werden können.

Stapelverarbeitung, Drosselung und Komprimierung

Aus Gründen der Effizienz sammelt der Produzent Stapel von Datensätzen, um sie an die Server zu senden. Wenn Sie die Komprimierung aktivieren, komprimiert der Producer jeden Stapel. Dies verbessert möglicherweise die Leistung, da weniger Daten über das Netz übertragen werden.

Wenn Sie versuchen, Nachrichten schneller zu veröffentlichen, als sie an einen Server gesendet werden können, puffert der Producer sie automatisch in Anforderungen für die Stapelverarbeitung. Der Producer verwaltet für jede Partition einen Puffer aus nicht gesendeten Datensätzen. Irgendwann kommt der Punkt, an dem selbst bei der Dosierung nicht mehr die gewünschte Rate erreicht werden kann.

Es gibt noch einen weiteren Faktor, der Auswirkungen hat. Um zu verhindern, dass einzelne Producer oder Consumer den Cluster überfluten, wendet Event Streams Größenbeschränkungen für den Durchsatz an. Die Rate, mit der jeder Producer Daten sendet, wird berechnet und jeder Producer, der versucht, seine Rate zu überschreiten, wird gedrosselt. Das Drosseln wird angewendet, indem das Senden von Antworten an den Producer leicht verzögert wird. Dies funktioniert ähnlich wie eine Bremse.

Weitere Informationen zum Durchsatz finden Sie unter Grenzwerte und Kontingente.

Zusammenfassend kann gesagt werden, dass der Datensatz bei der Veröffentlichung einer Nachricht zunächst im Producer in einen Puffer geschrieben wird. Der Producer erstellt im Hintergrund einen Stapel und sendet die Datensätze an den Server. Der Server antwortet dann dem Producer und wendet möglicherweise eine Drosselungsverzögerung an, wenn der Producer eine zu schnelle Veröffentlichung durchführt. Wenn der Puffer im Producer voll ist, wird der Sendeaufruf des Producers verzögert, kann aber letztendlich mit einer Ausnahme fehlschlagen.

Zustellungssemantik

Kafka bietet die folgenden verschiedenen Möglichkeiten der Nachrichtenzustellungssemantik:

• Höchstens einmal: Nachrichten können verloren gehen und werden nicht erneut zugestellt.
• Mindestens einmal: Nachrichten gehen nie verloren, aber es kann zu Duplikaten kommen.
• Genau einmal: Es gehen keine Nachrichten verloren und es gibt keine Duplikate.

Die Zustellungssemantik wird durch die folgenden Einstellungen bestimmt:

• acks
• retries
• enable.idempotence

Standardmäßig verwendet Kafka mindestens eine Semantik.

Um die "Exact Once"-Semantik zu aktivieren, müssen Sie die idempotenten oder transaktionalen Produzenten verwenden. Der idempotente Procucer wird aktiviert, indem für enable.idempotence die Einstellung true festgelegt wird. Damit wird sichergestellt, dass genau ein Exemplar jeder Nachricht in Kafka geschrieben wird, auch bei Wiederholungen. Der transaktionsorientierte Producer ermöglicht das Senden von Daten an mehrere Partitionen, sodass entweder alle Nachrichten erfolgreich zugestellt werden, oder keine der Nachrichten. D. h., eine Transaktion wird entweder vollständig festgeschrieben oder vollständig verworfen. Sie können auch Offsets in Transaktionen aufnehmen, um Anwendungen zu erstellen, die Nachrichten in Kafka lesen, verarbeiten und schreiben.

Code-Snippets

Diese Code-Snippets sind auf einer hohen Ebene, um die beteiligten Konzepte zu veranschaulichen. Vollständige Beispiele finden Sie in den Event Streams -Beispielen in GitHub-.

Um einen Konsumenten mit Event Streamszu verbinden, müssen Sie Serviceberechtigungsnachweise erstellen. Weitere Informationen finden Sie in Mit Event Streams verbinden.

Im Producer-Code müssen Sie zunächst die Gruppe der Konfigurationseigenschaften erstellen. Alle Verbindungen zu Event Streams sind durch TLS und Benutzer- und Passwortauthentifizierung gesichert, so dass Sie mindestens diese Eigenschaften benötigen. Ersetzen Sie BOOTSTRAP_ENDPOINTS, USER und PASSWORD durch die Werte Ihrer eigenen Dienstanmeldedaten:

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

Um Nachrichten zu senden, müssen Sie auch Serialisierer für die Schlüssel und Werte angeben, zum Beispiel:

 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Diese Serializer müssen den von den Konsumenten verwendeten Deserializern entsprechen.

Dann verwenden Sie einen KafkaProducer zum Senden von Nachrichten, wobei jede Nachricht durch einen ProducerRecord dargestellt wird. Vergessen Sie nicht, KafkaProducer zu schließen, wenn Sie fertig sind. Dieser Code sendet nur die Nachricht, wartet jedoch nicht ab, ob der Sendevorgang erfolgreich war. Die Nachricht wird an das Topic T1 mit der Zeichenfolge key und dem Wert value gesendet.

 Producer<String, String> producer = new KafkaProducer<>(props);
 producer.send(new ProducerRecord<String, String>("T1", "key", "value"));
 producer.close();

Die Methode send() ist asynchron und gibt ein Future zurück, mit dem sein Abschluss überprüft werden kann:

 Future<RecordMetadata> f = producer.send(new ProducerRecord<String, String>("T1", "key", "value"));
// Do some other stuff
// Now wait for the result of the send
 RecordMetadata rm = f.get();
 long offset = rm.offset;

Alternativ dazu können Sie beim Senden der Nachricht einen Rückruf angeben:

producer.send(new ProducerRecord<String,String>("T1","key","value", new Callback() {
          public void onCompletion(RecordMetadata metadata, Exception exception) {
                     // This is called when the send completes, either successfully or with an exception
          }
});

Weitere Informationen finden Sie in der Javadoc für den Kafka-Client.