Apache Kafka - ตัวอย่างกลุ่มผู้บริโภค

กลุ่มผู้บริโภคคือการใช้งานแบบหลายเธรดหรือหลายเครื่องจากหัวข้อ Kafka

กลุ่มผู้บริโภค

  • ผู้บริโภคสามารถเข้าร่วมกลุ่มได้โดยใช้group.idเดียวกัน

  • ความขนานสูงสุดของกลุ่มคือจำนวนผู้บริโภคในกลุ่ม←ไม่มีพาร์ติชัน

  • Kafka กำหนดพาร์ติชันของหัวข้อให้กับผู้บริโภคในกลุ่มเพื่อให้แต่ละพาร์ติชันถูกใช้โดยผู้บริโภคเพียงรายเดียวในกลุ่ม

  • Kafka รับประกันว่าข้อความจะถูกอ่านโดยผู้บริโภครายเดียวในกลุ่มเท่านั้น

  • ผู้บริโภคสามารถดูข้อความตามลำดับที่จัดเก็บไว้ในบันทึก

การปรับสมดุลของผู้บริโภคอีกครั้ง

การเพิ่มกระบวนการ / เธรดมากขึ้นจะทำให้ Kafka ปรับสมดุลใหม่ หากผู้บริโภคหรือนายหน้ารายใดไม่สามารถส่ง heartbeat ไปยัง ZooKeeper ก็สามารถกำหนดค่าใหม่ได้ผ่านคลัสเตอร์ Kafka ในระหว่างการปรับสมดุลใหม่ Kafka จะกำหนดพาร์ติชันที่พร้อมใช้งานให้กับเธรดที่มีอยู่โดยอาจย้ายพาร์ติชันไปยังกระบวนการอื่น

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 ConsumerGroup {
   public static void main(String[] args) throws Exception {
      if(args.length < 2){
         System.out.println("Usage: consumer <topic> <groupname>");
         return;
      }
      
      String topic = args[0].toString();
      String group = args[1].toString();
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", group);
      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.serialization.ByteArraySerializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
      
      consumer.subscribe(Arrays.asList(topic));
      System.out.println("Subscribed to topic " + topic);
      int i = 0;
         
      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
               System.out.printf("offset = %d, key = %s, value = %s\n", 
               record.offset(), record.key(), record.value());
      }     
   }  
}

การรวบรวม

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java

การดำเนินการ

>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group
>>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group

ที่นี่เราได้สร้างชื่อกลุ่มตัวอย่างเป็นmy-groupกับผู้บริโภคสองคน ในทำนองเดียวกันคุณสามารถสร้างกลุ่มและจำนวนผู้บริโภคในกลุ่มได้

อินพุต

เปิดผู้ผลิต CLI และส่งข้อความเช่น -

Test consumer group 01
Test consumer group 02

ผลลัพธ์ของกระบวนการแรก

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 01

ผลลัพธ์ของกระบวนการที่สอง

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 02

ตอนนี้หวังว่าคุณจะเข้าใจ SimpleConsumer และ ConsumeGroup โดยใช้การสาธิตไคลเอ็นต์ Java ตอนนี้คุณมีแนวคิดเกี่ยวกับวิธีการส่งและรับข้อความโดยใช้ไคลเอ็นต์ Java ให้เราดำเนินการรวม Kafka กับเทคโนโลยีข้อมูลขนาดใหญ่ในบทต่อไป