Apache Kafka - Einfaches Produzentenbeispiel

Lassen Sie uns eine Anwendung zum Veröffentlichen und Konsumieren von Nachrichten mit einem Java-Client erstellen. Der Kafka Producer Client besteht aus den folgenden APIs.

KafkaProducer API

Lassen Sie uns den wichtigsten Satz der Kafka-Produzenten-API in diesem Abschnitt verstehen. Der zentrale Teil der KafkaProducer-API ist die KafkaProducer- Klasse. Die KafkaProducer-Klasse bietet die Möglichkeit, einen Kafka-Broker in seinem Konstruktor mit den folgenden Methoden zu verbinden.

  • Die KafkaProducer-Klasse bietet eine Sendemethode zum asynchronen Senden von Nachrichten an ein Thema. Die Signatur von send () lautet wie folgt

producer.send(new ProducerRecord<byte[],byte[]>(topic, 
partition, key1, value1) , callback);

  • ProducerRecord - Der Produzent verwaltet einen Puffer mit Datensätzen, die darauf warten, gesendet zu werden.

  • Callback - Ein vom Benutzer angegebener Rückruf, der ausgeführt wird, wenn der Datensatz vom Server bestätigt wurde (null bedeutet, dass kein Rückruf erfolgt).

  • Die KafkaProducer-Klasse bietet eine Flush-Methode, um sicherzustellen, dass alle zuvor gesendeten Nachrichten tatsächlich abgeschlossen wurden. Die Syntax der Flush-Methode lautet wie folgt:

public void flush()

  • Die KafkaProducer-Klasse bietet die partitionFor-Methode, mit deren Hilfe die Partitionsmetadaten für ein bestimmtes Thema abgerufen werden können. Dies kann für die benutzerdefinierte Partitionierung verwendet werden. Die Signatur dieser Methode lautet wie folgt:

public Map metrics()

Es gibt die Karte der internen Metriken zurück, die vom Hersteller verwaltet werden.

  • public void close () - Die KafkaProducer-Klasse stellt enge Methodenblöcke bereit, bis alle zuvor gesendeten Anforderungen abgeschlossen sind.

Hersteller-API

Der zentrale Teil der Producer-API ist die Producer- Klasse. Die Producer-Klasse bietet eine Option zum Verbinden des Kafka-Brokers in seinem Konstruktor mit den folgenden Methoden.

Die Produzentenklasse

Die Producer-Klasse bietet die Sendemethode an send Nachrichten zu einzelnen oder mehreren Themen mit den folgenden Signaturen.

public void send(KeyedMessaget<k,v> message) 
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(List<KeyedMessage<k,v>>messages)
- sends data to multiple topics.
Properties prop = new Properties();
prop.put(producer.type,”async”)
ProducerConfig config = new ProducerConfig(prop);

Es gibt zwei Arten von Produzenten - Sync und Async.

Die gleiche API-Konfiguration gilt auch für Sync Producer. Der Unterschied zwischen ihnen besteht darin, dass ein Synchronisierungsproduzent Nachrichten direkt sendet, Nachrichten jedoch im Hintergrund. Async-Produzent wird bevorzugt, wenn Sie einen höheren Durchsatz wünschen. In früheren Versionen wie 0.8 hat ein asynchroner Produzent keinen Rückruf für send (), um Fehlerbehandlungsroutinen zu registrieren. Dies ist nur in der aktuellen Version von 0.9 verfügbar.

public void close ()

Produzentenklasse bietet close Methode zum Schließen der Produzentenpoolverbindungen zu allen Kafka-Brokern.

Konfigurationseinstellungen

Die Hauptkonfigurationseinstellungen der Producer-API sind zum besseren Verständnis in der folgenden Tabelle aufgeführt.

S.No. Konfigurationseinstellungen und Beschreibung
1

client.id

identifiziert die Herstelleranwendung
2

producer.type

entweder synchron oder asynchron
3

acks

Die acks-Konfiguration steuert die Kriterien, unter denen Herstelleranforderungen als vollständig betrachtet werden.
4

retries

Wenn die Herstelleranforderung fehlschlägt, versuchen Sie es automatisch mit einem bestimmten Wert.
5

bootstrap.servers

Bootstrapping-Liste der Broker.
6

linger.ms

Wenn Sie die Anzahl der Anforderungen reduzieren möchten, können Sie verweilen.ms auf einen Wert setzen, der größer als ein Wert ist.
7

key.serializer

Schlüssel für die Serializer-Schnittstelle.
8

value.serializer

Wert für die Serializer-Schnittstelle.
9

batch.size

Puffergröße.
10

buffer.memory

Steuert die Gesamtmenge an Speicher, die dem Produzenten zum Puffern zur Verfügung steht.

ProducerRecord API

ProducerRecord ist ein Schlüssel / Wert-Paar, das an den Kafka-Cluster gesendet wird. ProducerRecord-Klassenkonstruktor zum Erstellen eines Datensatzes mit Partitions-, Schlüssel- und Wertepaaren unter Verwendung der folgenden Signatur.

public ProducerRecord (string topic, int partition, k key, v value)

  • Topic - Benutzerdefinierter Themenname, der an die Aufzeichnung angehängt wird.

  • Partition - Anzahl der Partitionen

  • Key - Der Schlüssel, der in den Datensatz aufgenommen wird.

  • Value - Inhalt aufzeichnen

public ProducerRecord (string topic, k key, v value)

Der ProducerRecord-Klassenkonstruktor wird verwendet, um einen Datensatz mit Schlüssel-, Wertepaaren und ohne Partition zu erstellen.

  • Topic - Erstellen Sie ein Thema, um einen Datensatz zuzuweisen.

  • Key - Schlüssel für die Aufzeichnung.

  • Value - Inhalte aufzeichnen.

public ProducerRecord (string topic, v value)

Die ProducerRecord-Klasse erstellt einen Datensatz ohne Partition und Schlüssel.

  • Topic - ein Thema erstellen.

  • Value - Inhalte aufzeichnen.

Die ProducerRecord-Klassenmethoden sind in der folgenden Tabelle aufgeführt:

S.No. Klassenmethoden und Beschreibung
1

public string topic()

Das Thema wird an den Datensatz angehängt.
2

public K key()

Schlüssel, der in den Datensatz aufgenommen wird. Wenn kein solcher Schlüssel vorhanden ist, wird hier erneut null gesetzt.
3

public V value()

Inhalt aufzeichnen.
4

partition()

Partitionsanzahl für den Datensatz

SimpleProducer-Anwendung

Starten Sie vor dem Erstellen der Anwendung zunächst ZooKeeper und Kafka Broker und erstellen Sie dann mit dem Befehl create topic ein eigenes Thema in Kafka Broker. Erstellen Sie anschließend eine Java-Klasse mit dem Namen Sim-pleProducer.java und geben Sie die folgende Codierung ein.

//import util.properties packages
import java.util.Properties;

//import simple producer packages
import org.apache.kafka.clients.producer.Producer;

//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;

//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;

//Create java class named “SimpleProducer”
public class SimpleProducer {
   
   public static void main(String[] args) throws Exception{
      
      // Check arguments length value
      if(args.length == 0){
         System.out.println("Enter topic name”);
         return;
      }
      
      //Assign topicName to string variable
      String topicName = args[0].toString();
      
      // create instance for properties to access producer configs   
      Properties props = new Properties();
      
      //Assign localhost id
      props.put("bootstrap.servers", “localhost:9092");
      
      //Set acknowledgements for producer requests.      
      props.put("acks", “all");
      
      //If the request fails, the producer can automatically retry,
      props.put("retries", 0);
      
      //Specify buffer size in config
      props.put("batch.size", 16384);
      
      //Reduce the no of requests less than 0   
      props.put("linger.ms", 1);
      
      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
         
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      
      Producer<String, String> producer = new KafkaProducer
         <String, String>(props);
            
      for(int i = 0; i < 10; i++)
         producer.send(new ProducerRecord<String, String>(topicName, 
            Integer.toString(i), Integer.toString(i)));
               System.out.println(“Message sent successfully”);
               producer.close();
   }
}

Compilation - Die Anwendung kann mit dem folgenden Befehl kompiliert werden.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution - Die Anwendung kann mit dem folgenden Befehl ausgeführt werden.

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleProducer <topic-name>

Output

Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.
>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning
1
2
3
4
5
6
7
8
9
10

Einfaches Verbraucherbeispiel

Ab sofort haben wir einen Produzenten erstellt, der Nachrichten an den Kafka-Cluster sendet. Lassen Sie uns nun einen Konsumenten erstellen, der Nachrichten aus dem Kafka-Cluster konsumiert. Die KafkaConsumer-API wird verwendet, um Nachrichten aus dem Kafka-Cluster zu konsumieren. Der KafkaConsumer-Klassenkonstruktor ist unten definiert.

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

configs - Geben Sie eine Karte mit Verbraucherkonfigurationen zurück.

Die KafkaConsumer-Klasse verfügt über die folgenden wichtigen Methoden, die in der folgenden Tabelle aufgeführt sind.

S.No. Methode und Beschreibung
1

public java.util.Set<TopicPar-tition> assignment()

Rufen Sie die Partitionen ab, die derzeit vom Konsumenten zugewiesen wurden.
2

public string subscription()

Abonnieren Sie die angegebene Themenliste, um dynamisch signierte Partitionen zu erhalten.
3

public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener)

Abonnieren Sie die angegebene Themenliste, um dynamisch signierte Partitionen zu erhalten.
4

public void unsubscribe()

Deaktivieren Sie die Themen aus der angegebenen Liste der Partitionen.
5

public void sub-scribe(java.util.List<java.lang.String> topics)

Abonnieren Sie die angegebene Themenliste, um dynamisch signierte Partitionen zu erhalten. Wenn die angegebene Themenliste leer ist, wird sie genauso behandelt wie das Abbestellen ().
6

public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener)

Das Argumentmuster bezieht sich auf das Abonnementmuster im Format des regulären Ausdrucks, und das Listener-Argument erhält Benachrichtigungen vom Abonnementmuster.
7

public void as-sign(java.util.List<TopicParti-tion> partitions)

Weisen Sie dem Kunden manuell eine Liste von Partitionen zu.
8

poll()

Abrufen von Daten für die angegebenen Themen oder Partitionen mithilfe einer der Abonnement- / Zuweisungs-APIs. Dies gibt einen Fehler zurück, wenn die Themen vor dem Abrufen von Daten nicht abonniert wurden.
9

public void commitSync()

Festschreiben von Offsets, die bei der letzten Umfrage () für die gesamte unterzeichnete Liste von Themen und Partitionen zurückgegeben wurden. Die gleiche Operation wird auf commitAsyn () angewendet.
10

public void seek(TopicPartition partition, long offset)

Rufen Sie den aktuellen Versatzwert ab, den der Verbraucher bei der nächsten poll () -Methode verwendet.
11

public void resume()

Setzen Sie die angehaltenen Partitionen fort.
12

public void wakeup()

Wecken Sie den Verbraucher.

ConsumerRecord-API

Die ConsumerRecord-API wird zum Empfangen von Datensätzen vom Kafka-Cluster verwendet. Diese API besteht aus einem Themennamen, einer Partitionsnummer, von der der Datensatz empfangen wird, und einem Offset, der auf den Datensatz in einer Kafka-Partition verweist. Die ConsumerRecord-Klasse wird verwendet, um einen Consumer-Datensatz mit bestimmten Themennamen, Partitionsanzahl und <Schlüssel, Wert> -Paaren zu erstellen. Es hat die folgende Signatur.

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

  • Topic - Der Themenname für den vom Kafka-Cluster empfangenen Verbraucherdatensatz.

  • Partition - Partition für das Thema.

  • Key - Der Schlüssel des Datensatzes, wenn kein Schlüssel vorhanden ist, wird null zurückgegeben.

  • Value - Inhalt aufzeichnen.

ConsumerRecords-API

Die ConsumerRecords-API fungiert als Container für ConsumerRecord. Diese API wird verwendet, um die Liste von ConsumerRecord pro Partition für ein bestimmtes Thema zu speichern. Sein Konstruktor ist unten definiert.

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)

  • TopicPartition - Geben Sie eine Partitionskarte für ein bestimmtes Thema zurück.

  • Records - Rückgabeliste von ConsumerRecord.

In der ConsumerRecords-Klasse sind die folgenden Methoden definiert.

S.No. Methoden und Beschreibung
1

public int count()

Die Anzahl der Datensätze für alle Themen.
2

public Set partitions()

Der Satz von Partitionen mit Daten in diesem Datensatz (wenn keine Daten zurückgegeben wurden, ist der Satz leer).
3

public Iterator iterator()

Mit Iterator können Sie eine Sammlung durchlaufen, Elemente abrufen oder verschieben.
4

public List records()

Liste der Datensätze für die angegebene Partition abrufen.

Konfigurationseinstellungen

Die Konfigurationseinstellungen für die Hauptkonfigurationseinstellungen der Consumer-Client-API sind unten aufgeführt:

S.No. Einstellungen und Beschreibung
1

bootstrap.servers

Bootstrapping-Liste der Broker.
2

group.id

Weist einer Gruppe einen einzelnen Verbraucher zu.
3

enable.auto.commit

Aktivieren Sie die automatische Festschreibung für Offsets, wenn der Wert true ist, andernfalls nicht festgeschrieben.
4

auto.commit.interval.ms

Gibt zurück, wie oft aktualisierte verbrauchte Offsets in ZooKeeper geschrieben werden.
5

session.timeout.ms

Gibt an, wie viele Millisekunden Kafka darauf wartet, dass der ZooKeeper auf eine Anfrage (Lesen oder Schreiben) antwortet, bevor er aufgibt und weiterhin Nachrichten konsumiert.

SimpleConsumer-Anwendung

Die Herstelleranwendungsschritte bleiben hier gleich. Starten Sie zuerst Ihren ZooKeeper- und Kafka-Broker. Erstellen Sie dann eine SimpleConsumer- Anwendung mit der Java-Klasse SimpleCon-sumer.java und geben Sie den folgenden Code ein.

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {
   public static void main(String[] args) throws Exception {
      if(args.length == 0){
         System.out.println("Enter topic name");
         return;
      }
      //Kafka consumer configuration settings
      String topicName = args[0].toString();
      Properties props = new Properties();
      
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", "test");
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer
         <String, String>(props);
      
      //Kafka Consumer subscribes list of topics here.
      consumer.subscribe(Arrays.asList(topicName))
      
      //print the topic name
      System.out.println("Subscribed to topic " + topicName);
      int i = 0;
      
      while (true) {
         ConsumerRecords<String, String> records = con-sumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
         
         // print the offset,key and value for the consumer records.
         System.out.printf("offset = %d, key = %s, value = %s\n", 
            record.offset(), record.key(), record.value());
      }
   }
}

Compilation - Die Anwendung kann mit dem folgenden Befehl kompiliert werden.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution − Die Anwendung kann mit dem folgenden Befehl ausgeführt werden

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

Input- Öffnen Sie die Produzenten-CLI und senden Sie einige Nachrichten an das Thema. Sie können die kleine Eingabe als "Hallo Verbraucher" eingeben.

Output - Es folgt die Ausgabe.

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer
ja/tutorial
es/tutorial
de/tutorial
fr/tutorial
th/tutorial
pt/tutorial
ru/tutorial
vi/tutorial
it/tutorial
ko/tutorial
tr/tutorial
id/tutorial
pl/tutorial
hi/tutorial
japost
espost
depost
frpost
thpost
ptpost
rupost
vipost
itpost
kopost
trpost
idpost
plpost
hipost

© Copyright 2021 - 2024 | All Rights Reserved