Apache Kafka - Basit Üretici Örneği

Java istemcisi kullanarak mesajları yayınlamak ve tüketmek için bir uygulama oluşturalım. Kafka üretici istemcisi aşağıdaki API'lerden oluşur.

KafkaProducer API

Bu bölümdeki en önemli Kafka üretici API kümesini anlayalım. KafkaProducer API'nin merkezi kısmı, KafkaProducer sınıfıdır. KafkaProducer sınıfı, yapıcısındaki bir Kafka aracısını aşağıdaki yöntemlerle bağlama seçeneği sağlar.

• KafkaProducer sınıfı, iletileri bir konuya eşzamansız olarak göndermek için gönderme yöntemi sağlar. Send () imzası aşağıdaki gibidir

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

• ProducerRecord - Üretici, gönderilmeyi bekleyen bir kayıt tamponunu yönetir.

• Callback - Kayıt sunucu tarafından onaylandığında yürütülecek kullanıcı tarafından sağlanan bir geri arama (boş, geri arama olmadığını gösterir).

• KafkaProducer sınıfı, önceden gönderilen tüm iletilerin gerçekten tamamlandığından emin olmak için bir temizleme yöntemi sağlar. Flush yönteminin sözdizimi aşağıdaki gibidir -

public void flush()

• KafkaProducer sınıfı, belirli bir konu için bölüm meta verilerini elde etmeye yardımcı olan partitionFor yöntemini sağlar. Bu, özel bölümleme için kullanılabilir. Bu yöntemin imzası aşağıdaki gibidir -

public Map metrics()

Üretici tarafından tutulan dahili ölçümlerin haritasını döndürür.

• public void close () - KafkaProducer sınıfı, önceden gönderilen tüm istekler tamamlanıncaya kadar yakın yöntem blokları sağlar.

Üretici API'si

Producer API'nin merkezi kısmı, Üretici sınıfıdır. Üretici sınıfı, aşağıdaki yöntemlerle yapıcısındaki Kafka aracısını bağlama seçeneği sunar.

Üretici Sınıfı

Üretici sınıfı, send Aşağıdaki imzaları kullanarak tek veya birden çok konuya mesaj gönderin.

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);

İki tür üretici vardır - Sync ve Async.

Aynı API yapılandırması, Sync üreticisi için de geçerlidir . Aralarındaki fark, bir senkronizasyon üreticisinin mesajları doğrudan göndermesi, ancak mesajları arka planda göndermesidir. Daha yüksek bir iş hacmi istediğinizde zaman uyumsuz üretici tercih edilir. 0.8 gibi önceki sürümlerde, zaman uyumsuz bir üreticinin hata işleyicileri kaydetmek için send () için bir geri araması yoktur. Bu, yalnızca mevcut 0.9 sürümünde mevcuttur.

public void close ()

Üretici sınıfı sağlar close Üretici havuzu bağlantılarını tüm Kafka brokerlerine kapatma yöntemi.

Yapılandırma ayarları

Daha iyi anlaşılması için Producer API'nin ana yapılandırma ayarları aşağıdaki tabloda listelenmiştir -

S.No	Yapılandırma Ayarları ve Açıklama
1	client.id üretici uygulamasını tanımlar
2	producer.type eşzamanlı veya eşzamansız
3	acks Acks yapılandırması, üretici talepleri altındaki kriterlerin eksiksiz olarak kabul edilir.
4	retries Üretici talebi başarısız olursa, otomatik olarak belirli bir değerle yeniden deneyin.
5	bootstrap.servers aracıların önyükleme listesi.
6	linger.ms istek sayısını azaltmak istiyorsanız, linger.ms'yi belirli bir değerden daha büyük bir değere ayarlayabilirsiniz.
7	key.serializer Serileştirici arabirimi anahtarı.
8	value.serializer serileştirici arabirimi için değer.
9	batch.size Tampon boyutu.
10	buffer.memory üreticinin arabellekleme için kullanabileceği toplam bellek miktarını kontrol eder.

ProducerRecord API

ProducerRecord, aşağıdaki imzayı kullanarak bölüm, anahtar ve değer çiftleri ile bir kayıt oluşturmak için Kafka cluster.ProducerRecord sınıf yapıcısına gönderilen bir anahtar / değer çiftidir.

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

• Topic - kayda eklenecek kullanıcı tanımlı konu adı.

• Partition - bölüm sayısı

• Key - Kayda dahil edilecek anahtar.

• Value - İçeriği kaydedin

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

ProducerRecord sınıf yapıcısı, anahtarlı, değer çiftli ve bölümsüz bir kayıt oluşturmak için kullanılır.

• Topic - Kayıt atamak için bir konu oluşturun.

• Key - kaydın anahtarı.

• Value - içeriği kaydedin.

public ProducerRecord (string topic, v value)

ProducerRecord sınıfı, bölüm ve anahtarsız bir kayıt oluşturur.

• Topic - bir konu oluşturun.

• Value - içeriği kaydedin.

ProducerRecord sınıfı yöntemleri aşağıdaki tabloda listelenmiştir -

S.No	Sınıf Yöntemleri ve Açıklama
1	public string topic() Konu kayda eklenecek.
2	public K key() Kayda dahil edilecek anahtar. Böyle bir anahtar yoksa, null burada yeniden döndürülecektir.
3	public V value() İçeriği kaydedin.
4	partition() Kayıt için bölüm sayısı

SimpleProducer uygulaması

Uygulamayı oluşturmadan önce, önce ZooKeeper ve Kafka broker'ı başlatın, ardından create topic komutunu kullanarak Kafka broker'da kendi konunuzu oluşturun. Bundan sonra Sim-pleProducer.java adında bir java sınıfı oluşturun ve aşağıdaki kodlamayı yazın.

//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 - Uygulama aşağıdaki komut kullanılarak derlenebilir.

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

Execution - Uygulama aşağıdaki komut kullanılarak yürütülebilir.

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

Basit Tüketici Örneği

Şu an itibariyle Kafka kümesine mesaj gönderecek bir üretici oluşturduk. Şimdi Kafka kümesinden gelen mesajları tüketecek bir tüketici oluşturalım. KafkaConsumer API, Kafka kümesinden gelen mesajları tüketmek için kullanılır. KafkaConsumer sınıfı yapıcısı aşağıda tanımlanmıştır.

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

configs - Tüketici yapılandırmalarının bir haritasını döndürür.

KafkaConsumer sınıfı, aşağıdaki tabloda listelenen aşağıdaki önemli yöntemlere sahiptir.

S.No	Yöntem ve Açıklama
1	public java.util.Set<TopicPar-tition> assignment() Tüketici tarafından şu anda atanan bölüm kümesini alın.
2	public string subscription() Dinamik olarak imzalanmış bölümler almak için verilen konu listesine abone olun.
3	public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener) Dinamik olarak imzalanmış bölümler almak için verilen konu listesine abone olun.
4	public void unsubscribe() Verilen bölüm listesinden konuların aboneliğini kaldırın.
5	public void sub-scribe(java.util.List<java.lang.String> topics) Dinamik olarak imzalanmış bölümler almak için verilen konu listesine abone olun. Verilen konu listesi boşsa, unsubscribe () ile aynı şekilde değerlendirilir.
6	public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener) Argüman modeli, düzenli ifade formatındaki abone modeline atıfta bulunur ve dinleyici argümanı abone modelinden bildirimler alır.
7	public void as-sign(java.util.List<TopicParti-tion> partitions) Müşteriye manuel olarak bölüm listesi atayın.
8	poll() Abone olma / atama API'lerinden birini kullanarak belirtilen konular veya bölümler için verileri alın. Veriler için oylama yapılmadan önce konular abone olunmazsa bu hata verir.
9	public void commitSync() Konuların ve bölümlerin tüm alt karalanmış listesi için son ankette () döndürülen kaydetme uzaklıkları. Aynı işlem commitAsyn () için de uygulanır.
10	public void seek(TopicPartition partition, long offset) Tüketicinin bir sonraki anket () yönteminde kullanacağı mevcut göreli konum değerini getir.
11	public void resume() Duraklatılan bölümleri devam ettirin.
12	public void wakeup() Tüketiciyi uyandırın.

ConsumerRecord API

ConsumerRecord API, Kafka kümesinden kayıtları almak için kullanılır. Bu API, kaydın alındığı bir konu adı, bölüm numarası ve bir Kafka bölümündeki kayda işaret eden bir ofsetten oluşur. ConsumerRecord sınıfı, belirli konu adı, bölüm sayısı ve <anahtar, değer> çiftleriyle bir tüketici kaydı oluşturmak için kullanılır. Aşağıdaki imzaya sahiptir.

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

• Topic - Kafka kümesinden alınan tüketici kaydı için konu adı.

• Partition - Konu için bölüm.

• Key - Anahtar yoksa, kaydın anahtarı boş döndürülür.

• Value - İçeriği kaydedin.

ConsumerRecords API

ConsumerRecords API, ConsumerRecord için bir kapsayıcı görevi görür. Bu API, belirli bir konu için bölüm başına Tüketici Kaydı listesini tutmak için kullanılır. Yapıcısı aşağıda tanımlanmıştır.

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

• TopicPartition - Belirli bir konu için bir bölüm haritası döndürür.

• Records - ConsumerRecord'un iade listesi.

ConsumerRecords sınıfı aşağıdaki yöntemlere sahiptir.

S.No	Yöntemler ve Açıklama
1	public int count() Tüm konular için kayıt sayısı.
2	public Set partitions() Bu kayıt kümesindeki verilere sahip bölümler kümesi (hiçbir veri döndürülmediyse küme boştur).
3	public Iterator iterator() Yineleyici, bir koleksiyonda dolaşmanızı, öğeleri elde etmenizi veya yeniden hareket ettirmenizi sağlar.
4	public List records() Verilen bölüm için kayıtların listesini alın.

Yapılandırma ayarları

Tüketici istemcisi API ana yapılandırma ayarları için yapılandırma ayarları aşağıda listelenmiştir -

S.No	Ayarlar ve Açıklama
1	bootstrap.servers Aracıların önyükleme listesi.
2	group.id Bireysel bir tüketiciyi bir gruba atar.
3	enable.auto.commit Değer doğruysa, aksi takdirde taahhüt edilmemişse ofsetler için otomatik kesinlemeyi etkinleştirin.
4	auto.commit.interval.ms ZooKeeper'a ne sıklıkla güncellenen tüketilen ofsetlerin yazıldığını döndürür.
5	session.timeout.ms Mesajları bırakmadan ve tüketmeye devam etmeden önce Kafka'nın ZooKeeper'ın bir isteğe yanıt vermesini (okuma veya yazma) kaç milisaniye bekleyeceğini belirtir.

SimpleConsumer Uygulaması

Üretici başvuru adımları burada aynı kalır. İlk önce ZooKeeper ve Kafka brokerınızı başlatın. Sonra oluşturmak SimpleConsumer adlı java sınıfı ile uygulamayı SimpleCon-sumer.java ve aşağıdaki kodu yazın.

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 - Uygulama aşağıdaki komut kullanılarak derlenebilir.

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

Execution − Uygulama aşağıdaki komut kullanılarak yürütülebilir

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

Input- Üretici CLI'yi açın ve konuya bazı mesajlar gönderin. Smple girişini 'Merhaba Tüketici' olarak koyabilirsiniz.

Output - Aşağıdakiler çıktı olacaktır.

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer