java - 如何使用 Java 中的 Kafka 8.2 API 生成消息？

Question

我正在尝试使用 Java 中的 kafka API。我正在使用以下 Maven 依赖项:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.8.2.0</version>
</dependency>

我无法连接到远程 kafka 服务器。我将 kafka 'server.properties' 文件端口属性更改为端口 8080。我可以毫无问题地同时启动 zookeeper 和 kafka 服务器。我还可以使用 kafka 下载附带的控制台生产者和消费者应用程序。 (Scala 2.10 版本)

我正在使用以下客户端代码创建远程 KafkaProducer

Properties propsProducer = new Properties();

propsProducer.put("bootstrap.servers", "172.xx.xx.xxx:8080");
propsProducer.put("key.serializer", org.apache.kafka.common.serialization.ByteArraySerializer.class);
propsProducer.put("value.serializer", org.apache.kafka.common.serialization.ByteArraySerializer.class);
propsProducer.put("topic.metadata.refresh.interval.ms", "0");

KafkaProducer<byte[], byte[]> m_kafkaProducer = new KafkaProducer<byte[], byte[]>(propsProducer);

创建生产者后，我可以运行以下行并获取返回的有效主题信息，前提是 strTopic 是现有主题名称。

List<PartitionInfo> partitionInfo = m_kafkaProducer.partitionsFor(strTopic);

当我尝试发送消息时，我会执行以下操作:

ProducerRecord<byte[], byte[]> prMessage = new ProducerRecord<byte[],byte[]>(strTopic, strMessage.getBytes());

RecordMetadata futureData = m_kafkaProducer.send(prMessage).get();

对 send() 的调用无限期地阻塞，当我手动终止进程时，我看到 ERROR Closing socket because error on kafka server(IOException, Connection Reset by Peer) 错误。

此外，host.name、advertised.host.name 和 advertised.port 属性仍然在“server.properties”文件中被注释掉，这毫无意义。哦，如果我换行:

propsProducer.put("bootstrap.servers", "172.xx.xx.xxx:8080");

到

propsProducer.put("bootstrap.servers", "127.0.0.1:8080");

并在安装了 kafka 服务器的同一台服务器上运行它，它可以工作，但我正在尝试远程使用它。

感谢任何帮助，如果我能澄清一点，请告诉我。

Answer 1

经过大量挖掘，我决定实现此处找到的示例:Kafka Producer Example .我缩短了代码并且没有实现分区程序类。我用列出的依赖项更新了我的 pom，但我仍然遇到同样的问题。最终，我进行了一些配置更改并且一切正常。

难题的最后一部分是在服务器和客户端计算机的/etc/hosts 中定义 Kafka 服务器。我在两个文件中都添加了以下内容。

172.xx.xx.xxx     serverHost1

同样，x 只是掩码。然后，我将 server.properties 文件中的 advertised.host.name 设置为 serverHost1。注意:我在服务器计算机上运行 ifconfig 后获得了该 IP。

我换了行

propsProducer.put("metadata.broker.list", "172.xx.xx.xxx:8080");

到

propsProducer.put("metadata.broker.list", "serverHost1:8080");

Kafka API 不喜欢我将 IP 定义为字符串这一事实。相反，它是从 etc/hosts 文件中查找 IP，尽管文档说:

“代理将向生产者和消费者通告的主机名。如果未设置，它将使用配置的“host.name”的值。否则，它将使用从 java.net.InetAddress.getCanonicalHostName() 返回的值。 "

这将只返回 IP，以字符串形式，如果未在客户端计算机的 etc/hosts 中定义，我之前使用，否则它返回与 IP 配对的名称(在我的例子中为 serverHost1)。另外，我也从未设置过 host.name 的值。