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 กับเทคโนโลยีข้อมูลขนาดใหญ่ในบทต่อไป