Consumer-Eigenschaften konfigurieren

Es gibt viele Konfigurationseinstellungen für den Verbraucher, die Aspekte seines Verhaltens steuern. Die folgenden Einstellungen sind einige der wichtigsten.

Es sind viele weitere Konfigurationseinstellungen verfügbar, aber lesen Sie die Dokumentation zuApache Kafka, bevor Sie mit dem Experimentieren beginnen.

Konsumentengruppen

Eine Konsumentengruppe ist eine Gruppe von Konsumenten, die kooperiert, um Nachrichten aus einem oder mehreren Themen zu konsumieren. Die Consumer in einer Gruppe verwenden alle denselben Wert für die group.id-Konfiguration. Wenn Sie mehr als einen Consumer zum Verarbeiten Ihrer Workload brauchen, können Sie mehrere Consumer in derselben Consumergruppe ausführen. Selbst wenn Sie nur einen Konsumenten benötigen, ist es üblich, auch einen Wert für group.idanzugeben.

Jede Konsumentengruppe verfügt über einen Server im Cluster, der als Koordinator bezeichnet wird, der den Konsumenten in der Gruppe Partitionen zuordnet. Diese Verantwortung wird auf die Server im Cluster verteilt, um die Last gleichmäßig zu verteilen. Die Zuordnung von Partitionen zu Consumern kann sich bei jedem neuen Ausgleich in der Gruppe ändern.

Wenn ein Consumer einer Consumergruppe beitritt, erkennt er den Koordinator für diese Gruppe. Der Konsument teilt dem Koordinator mit, dass er der Gruppe beitreten möchte, und der Koordinator startet eine Neuverteilung der Partitionen in der Gruppe, um das neue Member aufzunehmen.

Die Nachrichten aus einer einzelnen Partition werden durch jeweils nur einen Consumer in jeder Gruppe verarbeitet. Dieser Mechanismus stellt sicher, dass die Nachrichten in jeder Partition in der richtigen Reihenfolge verarbeitet werden. Das folgende Diagramm zeigt ein Beispiel, in dem ein Topic drei Partitionen und eine Consumergruppe, die dieses Topic konsumiert, zwei Consumer enthält. Einem Konsumenten in der Gruppe werden zwei Partitionen und dem anderen Konsumenten eine Partition zugeordnet.

Wenn eine der folgenden Änderungen in einer Konsumentengruppe auftritt, wird die Gruppe neu verteilt, indem die Zuordnung von Partitionen zu den Gruppenmitgliedern verschoben wird, um die Änderung aufzunehmen:

• Ein Konsument tritt der Gruppe bei.
• Ein Konsument verlässt die Gruppe.
• Ein Konsument wird vom Koordinator als nicht mehr live betrachtet.
• Einem vorhandenen Topic werden neue Partitionen hinzugefügt.

Wenn Sie eine Verbrauchergruppe haben, die neu ausbalanciert wird, beachten Sie, dass jeder Verbraucher, der die Gruppe verlassen hat, seine Commits zurückweist, bis er der Gruppe wieder beitritt. In diesem Fall muss der Consumer der Gruppe erneut beitreten. Möglicherweise wird ihm dann eine andere Partition zugewiesen.

Consumer-Aktivität

Kafka erkennt fehlgeschlagene Consumer automatisch, sodass aktiven Consumern Partitionen erneut zugeordnet werden können. Dabei werden zwei Mechanismen verwendet: Polling und Austausch von Überwachungssignalen.

Wenn der Stapel von Nachrichten, die von Consumer.poll(...) zurückgegeben werden, groß ist oder die Verarbeitung zeitaufwendig ist, kann die Verzögerung bis zum poll() erneuten Aufruf erheblich oder unvorhersehbar sein. In einigen Fällen muss ein langes maximales Abfrageintervall konfiguriert werden, damit die Consumer nicht aus ihren Gruppen entfernt werden, wenn die Nachrichtenverarbeitung längere Zeit dauert. Wenn dieser Mechanismus der einzige verfügbare Mechanismus ist, ist die Zeit, die zum Erkennen eines fehlgeschlagenen Konsumenten benötigt wird, ebenfalls lang.

Damit die Aktivität der Consumer einfacher gehandhabt werden kann, wurde in Kafka 0.10.1 das Heartbeating im Hintergrund hinzugefügt. Der Gruppenkoordinator erwartet von den Gruppenmitgliedern, dass sie regelmäßig Heartbeats senden, um anzuzeigen, dass sie weiterhin aktiv sind. Ein Überwachungssignalhintergrundthread wird im Konsumenten ausgeführt und sendet regelmäßige Überwachungssignale an den Koordinator. Wenn der Koordinator kein Überwachungssignal von einem Gruppenmitglied in _Sitzungszeitlimit_empfängt, entfernt er das Mitglied aus der Gruppe und startet eine Neuverteilung der Gruppe. Das Sitzungs-Timeout kann viel kürzer sein als das maximale Polling-Intervall, so dass die Zeit, die benötigt wird, um einen ausgefallenen Verbraucher zu erkennen, kurz sein kann, auch wenn die Nachrichtenverarbeitung lange dauert.

Sie können das maximale Abfrageintervall mit der Eigenschaft max.poll.interval.ms und das Sitzungszeitlimit mit der Eigenschaft session.timeout.ms konfigurieren. Sie müssen diese Einstellungen nur verwenden, wenn die Verarbeitung eines Nachrichtenstapels länger als 5 Minuten dauert.

Offsets verwalten

Für jede Konsumentengruppe verwaltet Kafka den festgeschriebenen Offset für jede verarbeitete Partition. Wenn ein Consumer eine Nachricht verarbeitet, wird die Nachricht nicht von der Partition entfernt. Stattdessen wird nur der aktuelle Offset aktualisiert, indem ein Prozess verwendet wird, der als Festschreiben des Offsets bezeichnet wird.

Event Streams speichert Consumer-Offset-Informationen für einen Zeitraum von 7 Tagen.

Die Geschwindigkeit der Nachrichtenverarbeitung steuern

Wenn Sie Probleme mit der Nachrichtenverarbeitung haben, die durch Nachrichtenüberflutung verursacht werden, können Sie eine Verbraucheroption einstellen, um die Geschwindigkeit des Nachrichtenverbrauchs zu steuern. Verwenden Sie fetch.max.bytes und max.poll.records, um zu steuern, wie viele Daten ein Aufruf von poll() zurückgeben kann.

Consumer-Neuverteilung handhaben

Wenn Verbraucher zu einer Gruppe hinzugefügt oder aus ihr entfernt werden, findet eine Neuverteilung der Gruppe statt, und die Verbraucher können keine Nachrichten konsumieren. Dies führt dazu, dass alle Verbraucher einer Verbrauchergruppe für einen kurzen Zeitraum nicht erreichbar sind.

Wenn Sie mit dem "on partitions revoked"-Callback benachrichtigt werden, verwenden Sie einen ConsumerRebalanceListener, um Offsets manuell zu bestätigen (wenn Sie nicht auto-commit verwenden) und die weitere Verarbeitung anzuhalten, bis Sie mit dem "on partition assigned"-Callback über den erfolgreichen Rebalance informiert werden.

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. Informationen zum Abrufen dieser Berechtigungsnachweise finden Sie unter Verbindung zu Event Streams.

Im Konsumentencode müssen Sie zunächst die Gruppe der Konfigurationseigenschaften erstellen. Alle Verbindungen zu Event Streams sind durch TLS und Benutzer-Passwort-Authentifizierung 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 konsumieren, müssen Sie auch Deserialisierer für die Schlüssel und Werte angeben, wie im folgenden Beispiel.

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Diese Deserializer müssen mit den von den Produzenten verwendeten Serializern übereinstimmen.

Verwenden Sie dann einen KafkaConsumer, um Nachrichten zu verarbeiten, wobei jede Nachricht durch einen Consumer-Datensatz (ConsumerRecord) dargestellt wird. Die gängigste Art, Nachrichten zu verarbeiten, ist es, den Consumer in eine Consumergruppe zu platzieren, indem Sie die Gruppen-ID festlegen und dann subscribe() für eine Liste von Topics aufrufen. Dem Konsumenten werden einige zu konsumierende Partitionen zugeordnet. Wenn jedoch mehr Konsumenten in der Gruppe als Partitionen im Topic vorhanden sind, werden dem Konsumenten möglicherweise keine Partitionen zugeordnet. Rufen Sie dann poll() in einer Schleife auf und empfangen Sie einen Stapel zu verarbeitender Nachrichten, wobei jede Nachricht durch einen Consumer-Datensatz (ConsumerRecord) dargestellt wird.

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
while (true) {
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records)
     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

Diese Consumer-Schleife läuft ewig, kann aber von einem anderen Thread aus durch den Aufruf von Consumer.wakeup() unterbrochen werden, um ein sauberes Herunterfahren zu erreichen.

Um Offsets per Commit manuell festzuschreiben, ist es zunächst notwendig, die Konfiguration enable.auto.commit auf false festzulegen. Verwenden Sie dann entweder Consumer.commmitSync() oder Consumer.commitAsync(), um den festgeschriebenen Offset laufend zu aktualisieren. Aus Gründen der Einfachheit verarbeitet dieses Beispiel die Datensätze für jede Partition und schreibt separat den letzten Offset fest.

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

Ausnahmebehandlung

Jede stabile Anwendung, die den Kafka-Client verwendet, muss Ausnahmen für bestimmte erwartete Situationen behandeln können. In einigen Fällen werden die Ausnahmen nicht direkt ausgelöst, weil einige Methoden asynchron sind und ihre Ergebnisse mithilfe eines Future oder eines Callbacks liefern. Sehen Sie sich Beispielcode in GitHub an, der vollständige Beispiele zeigt.

Bearbeiten Sie die folgende Liste von Ausnahmebedingungen in Ihrem Code: