java - 无法使用kafka Producer API与kafka服务器通信

Question

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

下面是我的 Producer 类

    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)

Answer 1

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