4

私は数日間Apache Kafkaで遊んでいますが、これが私の問題です.Webサイトの「クイックスタート」セクションに記載されているローカルテストをセットアップすると、すべて問題ありません.kafkaプロデューサー/コンシューマー、zookeeperサーバーそしてカフカブローカーは完璧に動作します。

ここで、リモート サーバー (node2 と呼びましょう) で実行する場合: - Zookeeper - ポート 2181 - Kafka Broker - ポート 9092 - kafka consumer

そして、ローカルコンピューターから実行すると: - kafka プロデューサー

node2 にファイアウォールがないと仮定します。接続はタイムアウトで終了します。

エラーログは次のとおりです。

/etc/java/jdk1.6.0_41/bin/java -Didea.launcher.port=7533 -Didea.launcher.bin.path=/home/kevin/Documents/idea-IU-123.169/bin -Dfile.encoding=UTF-8 -classpath /etc/java/jdk1.6.0_41/lib/dt.jar:/etc/java/jdk1.6.0_41/lib/tools.jar:/etc/java/jdk1.6.0_41/lib/jconsole.jar:/etc/java/jdk1.6.0_41/lib/htmlconverter.jar:/etc/java/jdk1.6.0_41/lib/sa-jdi.jar:/home/kevin/Desktop/kafka-0.7.2/examples/target/scala_2.8.0/classes:/home/kevin/Desktop/kafka-0.7.2/project/boot/scala-2.8.0/lib/scala-compiler.jar:/home/kevin/Desktop/kafka-0.7.2/project/boot/scala-2.8.0/lib/scala-library.jar:/home/kevin/Desktop/kafka-0.7.2/core/target/scala_2.8.0/classes:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/jopt-simple-3.2.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/log4j-1.2.15.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/zookeeper-3.3.4.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/zkclient-0.1.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/snappy-java-1.0.4.1.jar:/home/kevin/Desktop/kafka-0.7.2/examples/lib_managed/scala_2.8.0/compile/jopt-simple-3.2.jar:/home/kevin/Desktop/kafka-0.7.2/examples/lib_managed/scala_2.8.0/compile/log4j-1.2.15.jar:/home/kevin/Documents/idea-IU-123.169/lib/idea_rt.jar com.intellij.rt.execution.application.AppMain kafka.examples.KafkaConsumerProducerDemo
log4j:WARN No appenders could be found for logger (org.I0Itec.zkclient.ZkConnection).
log4j:WARN Please initialize the log4j system properly.
Exception in thread "Thread-0" java.net.ConnectException: Connection timed out
    at sun.nio.ch.Net.connect(Native Method)
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:532)
    at kafka.producer.SyncProducer.connect(SyncProducer.scala:173)
    at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:196)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:92)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:125)
    at kafka.producer.ProducerPool$$anonfun$send$1.apply$mcVI$sp(ProducerPool.scala:114)
    at kafka.producer.ProducerPool$$anonfun$send$1.apply(ProducerPool.scala:100)
    at kafka.producer.ProducerPool$$anonfun$send$1.apply(ProducerPool.scala:100)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:57)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:43)
    at kafka.producer.ProducerPool.send(ProducerPool.scala:100)
    at kafka.producer.Producer.zkSend(Producer.scala:137)
    at kafka.producer.Producer.send(Producer.scala:99)
    at kafka.javaapi.producer.Producer.send(Producer.scala:103)
    at kafka.examples.Producer.run(Producer.java:53)

Process finished with exit code 0

そして、ここに私のプロデューサーのコードがあります:

import java.util.Properties;
import kafka.javaapi.producer.ProducerData;
import kafka.producer.ProducerConfig;


public class Producer extends Thread{

  private final kafka.javaapi.producer.Producer<String, String> producer;
  private final String topic;
  private final Properties props = new Properties();

  public Producer(String topic)
  {
    props.put("zk.connect", "node2:2181");
    props.put("connect.timeout.ms", "5000");
    props.put("socket.timeout.ms", "30000");
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("producer.type", "sync");
    props.put("conpression.codec", "0");
    producer = new kafka.javaapi.producer.Producer<String, String>(new ProducerConfig(props));
    this.topic = topic;
  }

  public void run() {
      String messageStr = new String("Message_test");
      producer.send(new ProducerData<String, String>(topic, messageStr));
  }
}

**なので、切り替えもテストしました

props.put("zk.connect", "node2:2181");

に 

props.put("broker.list", "0:node2:9082");

その場合、正常に接続できます。**

4

1 に答える 1

3

http://kafka.apache.org/faq.htmlの項目 3 を参照してください。

回避策は、Kafka の server.properties でホスト名プロパティを明示的に設定することです。

これは、Zookeeper を使用して確認できます。kafka 0.7* を使用している場合は、ZkCli コンソールを開いて /brokers/ids/0 を取得すると、すべてのブローカーのメタデータを取得する必要があります。ここの IP アドレス/ホスト名が、プロデューサー コードで使用している Zk 接続文字列と一致していることを確認してください -

props.put("zk.connect", "node2:2181");

私の場合、ubuntu VM (同じボックス、異なる IP) に接続しているローカル マシンで実行されているプロデューサーを使用していましたが、この回避策が役に立ちました。

于 2013-04-17T22:09:55.573 に答える