Apache-kafka-consumer-group-example

Apache Kafka-消費者グループの例

消費者グループは、Kafkaトピックのマルチスレッドまたはマルチマシンの消費です。

消費者グループ

• 消費者は同じ「 group.id.」を使用してグループに参加できます
• グループの最大並列性は、グループ内のコンシューマの数←パーティションの数です。
• Kafkaは、トピックのパーティションをグループ内のコンシューマーに割り当てます。これにより、各パーティションは、グループ内の1つのコンシューマーのみによって消費されます。
• Kafkaは、グループ内の1人の消費者のみがメッセージを読むことを保証します。
• 消費者は、ログに保存された順序でメッセージを見ることができます。

消費者のリバランス

さらにプロセス/スレッドを追加すると、Kafkaのバランスが再調整されます。 消費者またはブローカーが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 " &plus; 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

ここでは、2つのコンシューマを持つ「 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

2番目のプロセスの出力

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

Javaクライアントデモを使用してSimpleConsumerとConsumeGroupを理解できたと思います。 これで、Javaクライアントを使用してメッセージを送受信する方法がわかりました。 次の章で、ビッグデータテクノロジーとのKafkaの統合を続けましょう。