6

我已经在单个节点上设置了 kafka 并启动了 zookeeper 和 kafka 服务器。我在控制台上为内部生产者和消费者测试了它，它运行良好。但是当我在控制台上运行内部 kafka 消费者和我的自定义生产者时不起作用。

下面是我的生产者类

    Properties props = new Properties();

    props.put("metadata.broker.list", "xx.xx.xx.xx:9092");
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("partitioner.class", "com.example.producer.SimplePartitioner");
    props.put("request.required.acks", "1");

    ProducerConfig config = new ProducerConfig(props);

    Producer<String, String> producer = new Producer<String, String>(config);
    KeyedMessage<String, String> data = new KeyedMessage<String, String>(
            "mails", "xxxx");
    producer.send(data);

当控件到达 producer.send() 时，它会在 3 次尝试后停止，但出现以下异常

java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
at kafka.producer.async.DefaultEventHandler$$anonfun$handle$2.apply$mcV$sp(DefaultEventHandler.scala:78)
at kafka.utils.Utils$.swallow(Utils.scala:172)
at kafka.utils.Logging$class.swallowError(Logging.scala:106)
at kafka.utils.Utils$.swallowError(Utils.scala:45)
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
at kafka.producer.Producer.send(Producer.scala:77)
at kafka.javaapi.producer.Producer.send(Producer.scala:33)
4

1 回答 1

3

我试图从外部 VM 的 Eclipse 中的 Producer 类连接到 kafka 服务器。我不得不在 kafka 的 config/ 中的 producer.properties 中将 localhost 替换为 ip 地址。

于 2015-03-17T13:25:07.550 回答