シンプルなカフカ消費者がメッセージを受け取っていない

Question

私はカフカの初心者で、簡単なカフカの消費者/生産者の例をKafkaConsumerとKafkaProducerで実行しています。コンシューマーが端末からコンシューマーを実行しているとき、コンシューマーはメッセージを受信して​​いますが、Javaコードを使用して聞くことはできません。 StackoverFlowでも同様の問題を検索しましたが（Links：Link1、Link2）、その解決策を試しましたが、何も私にとってうまくいかないようです。 カフカバージョン：kafka_2.10-0.10.2.1および対応する依存関係がpomで使用されています。

生産者と消費者のためのJavaコード：

public class SimpleProducer { 
public static void main(String[] args) throws InterruptedException { 
    Properties props = new Properties(); 
    props.put("bootstrap.servers", "localhost:9094"); 
    props.put("acks", "all"); 
    props.put("retries", 0); 
    props.put("batch.size", 16384); 
    props.put("linger.ms", 1); 
    props.put("buffer.memory", 33554432); 
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 

    Producer<String, String> producer = new KafkaProducer<>(props); 
    for(int i = 0; i < 10; i++) 
     producer.send(new ProducerRecord<String, String>("topic3", Integer.toString(i), Integer.toString(i))); 

    producer.close(); 

}} 

public class SimpleConsumer { 

public static void main(String[] args) { 
    Properties props = new Properties(); 
    props.put("bootstrap.servers", "localhost:9094"); 
    props.put("group.id", "test"); 
    props.put("zookeeper.connect", "localhost:2181"); 
    props.put("enable.auto.commit", "true"); 
    props.put("auto.commit.interval.ms", "1000"); 
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
    consumer.subscribe(Arrays.asList("topic3", "topic2")); 
    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()); 
    } 
}} 

出発カフカ：すべてのグループが使用して使用可能な bin/kafka-server-start.sh config/server.properties（私はすでにポートを設定して、プロパティでbrokeridファイル）

Answer 1

最初のチェック：

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 

次に、あなたのトピックは、CMDの下に使用して所属するグループを確認する：

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group <your group name> --describe 

あなたのトピックと関連するグループ名を見つけたら（そうでないグループをデフォルトに属している場合だけで、あなたのグループでgroup.idを置き換える）、その後小道具以下で試してみて、それが動作するかどうか私に教えて：

props.put("bootstrap.servers", "localhost:9092"); 
    props.put("group.id", "test-consumer-group"); // default topic name 
    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.StringDeserializer"); 
    props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); 
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); 

    //Kafka Consumer subscribes list of topics here. 
    consumer.subscribe(Arrays.asList(topicName)); // replace you topic name 

    //print the topic name 

    java.util.Map<String,java.util.List<PartitionInfo>> listTopics = consumer.listTopics(); 
    System.out.println("list of topic size :" + listTopics.size()); 

    for(String topic : listTopics.keySet()){ 
     System.out.println("topic name :"+topic); 
    } 

Answer 2

消費者がグループコーディネータに最初に登録するように、プロデューサを実行する前にコンシューマを実行します。その後、消費者がメッセージを消費するプロデューサを実行します。初めてコンシューマをグループコーディネータ消費者がメッセージを消費したオフセットがこれを使用するまで見つけるために注文するkafka-consumer-offset-checker.bat --group group-1 --topic testing-1 --zookeeper localhost:2181これは、消費者がトピックのどのオフセットを最後に消費したかを示します。