Was ist die richtige Reihenfolge beim Erstellen von Kafka Producer und Aufruf der Methoden send(), flush() und close()?

Question

Ich habe die unten genannten verschiedenen Modelle, wo ich Kafka-Produzent erstellen und verschiedene Methoden aufrufen, aber nicht sicher, was der richtige Ansatz ist, es zu programmieren, so dass der Fluss nicht brechen sollte und die Leistung nicht beeinträchtigt werden sollte. Freundlich helfen.

Modell 1:

for(int i=1; i < 100; i++){ 
    Producer<String, String> producer = new KafkaProducer<String, String>(props); 

    ProducerRecord<String, String> data = new ProducerRecord<String, String>(
     topicName, 
     String.valueOf(i) 
    ); 

    producer.send(data); 
    producer.close(); 
} 

Modell 2:

Producer<String, String> producer = new KafkaProducer<String, String>(props); 
for(int i=1; i < 100; i++) { 

    ProducerRecord<String, String> data = new ProducerRecord<String, String>(
     topicName, 
     String.valueOf(i) 
    ); 

    producer.send(data); 
    producer.close(); 
} 

Modell 3:

Producer<String, String> producer = new KafkaProducer<String, String>(props); 
for(int i=1; i < 100; i++){ 

    ProducerRecord<String, String> data = new ProducerRecord<String, String>(
     topicName, 
     String.valueOf(i) 
    ); 

    producer.send(data); 
} 
producer.close(); 

Modell 4:

for(int i=1; i < 100; i++){ 
    Producer<String, String> producer = new KafkaProducer<String, String>(props); 

    ProducerRecord<String, String> data = new ProducerRecord<String, String>(
     topicName, 
     String.valueOf(i) 
    ); 

    producer.send(data); 
    producer.flush(); 
    producer.close(); 
} 

Modell 5:

Producer<String, String> producer = new KafkaProducer<String, String>(props); 
for(int i=1; i < 100; i++){ 
    ProducerRecord<String, String> data = new ProducerRecord<String, String>(
     topicName, 
     String.valueOf(i) 
    ); 

    producer.send(data); 
    producer.flush(); 
    producer.close(); 
} 

Modell 6:

Producer<String, String> producer = new KafkaProducer<String, String>(props); 
for(int i=1; i < 100; i++){ 

    ProducerRecord<String, String> data = new ProducerRecord<String, String>(
     topicName, 
     String.valueOf(i) 
    ); 

    producer.send(data); 
    producer.flush(); 
} 
producer.close(); 

Answer 1

Das Modell 3 scheint soll

Producer<String, String> producer = new KafkaProducer<String, String>(props); 
    try { 
     for (int i = 1; i < 100; i++) { 
      ProducerRecord<String, String> data = new ProducerRecord<String, String>(topicName, String.valueOf(i)); 
      producer.send(data); 
     } 
    } catch (Exception e) { 
     e.printStackTrace(); 
    } finally { 
     producer.close(); 
    } 

Answer 2

Sie das Beispiel korrekt sein unten mit folgenden Änderung verwenden kann:

Properties props = new Properties(); 
props.put("batch.size", 16384); 
props.put("buffer.memory", 33554432); 
Producer<String, String> producer = new KafkaProducer<>(props); 
for (int i = 0; i < 100; i++) 
    producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i))); 

producer.close(); 

als bekannt, The send() -Methode ist asynchron. Wenn sie aufgerufen wird, fügt sie den Datensatz zu einem Puffer von ausstehenden Datensatz-Sends hinzu und kehrt sofort zurück. Dies ermöglicht dem Hersteller, einzelne Datensätze effizient zusammenzufassen.

Und wir können die buffer.memory oder batch.size automatisch