1

私は単純なカフカ プロデューサー コンシューマー クライアント API を試しています。コンソールからコンシューマーにメッセージが表示されるため、私のプロデューサー クラスは正常に動作しますが、コンシューマー コードを実行しているときに何も表示されず、何が問題なのか、どこにあるのかわかりません。間違いをしている

これはプロデューサーコードです -

package com.app.test;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class SimpleProducer {
    private static Producer<Integer, String> producer;
    private final Properties props = new Properties();public    SimpleProducer()
    {
      props.put("metadata.broker.list", "localhost:9092");
      props.put("serializer.class", "kafka.serializer.StringEncoder");
      props.put("request.required.acks", "1");
      producer = new Producer<Integer, String>(new ProducerConfig(props));
    }
public static void main(String[] args) {
    SimpleProducer sp = new SimpleProducer();
    String topic = "test4";
    String messageStr = "hello";
    KeyedMessage<Integer, String> data = new KeyedMessage<Integer, String>(topic, messageStr);
    System.out.println("producer : "+producer);
    producer.send(data);
    producer.close();
  }
}

消費者クラスは -

package com.app.test;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class SimpleHLConsumer {

  private final ConsumerConnector consumer;
  private final String topic;

  public SimpleHLConsumer(String zookeeper, String groupId, String topic) {
      Properties props = new Properties();
    props.put("zookeeper.connect", zookeeper);
    props.put("group.id", groupId);
    props.put("zookeeper.session.timeout.ms", "500");
    props.put("zookeeper.sync.time.ms", "250");
    props.put("auto.commit.interval.ms", "1000");

    consumer = Consumer.createJavaConsumerConnector(
    new ConsumerConfig(props));
    this.topic = topic;
  }



  public void testConsumer() {


    Map<String, Integer> topicCount = new HashMap<String, Integer>();
        // Define single thread for topic
    topicCount.put(topic, new Integer(1));      

    System.out.println("check1");

    Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumer.createMessageStreams(topicCount);

    System.out.println("check2");
    List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(topic);

   for (KafkaStream stream : streams) {
        System.out.println("test----");
        System.out.println("test----"+stream.toString());
        ConsumerIterator<byte[], byte[]> consumerIte = stream.iterator();
        while (consumerIte.hasNext()) {
            try {
                System.out.println("Message from Single Topic: " + new String(consumerIte.next().message(), "UTF-8"));
            } catch (UnsupportedEncodingException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    if (consumer != null)
      consumer.shutdown();
  }



  public static void main(String[] args) {

    String topic = "test4";
SimpleHLConsumer simpleHLConsumer = new SimpleHLConsumer("localhost:2181", "testgroup", topic);
    simpleHLConsumer.testConsumer();
  }
}

確認のために、testConsumer() メソッドで sysout を使用して 2 つのチェックを適用したため、実行中に check1 のみが表示されます。つまり、コードが check2 に到達していません。何らかの問題があると思います。consumer.createMessageStreams(topicCount);その理由と解決方法を教えてください。 ?

4

1 に答える 1

1

コードに問題はありません。

jar にkafka-version (使用している kafka のバージョン)を含む jar をビルドして実行し、プロデューサー コンソールからメッセージを送信してみてください。

これがあなたを助けることを願っています。

于 2015-03-26T14:53:00.647 に答える