如何在 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。我缩短了代码并且没有实现分区器 class。我用列出的依赖项更新了我的 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，尽管文档说：

"Hostname the broker will advertise to producers and consumers. If not set, it uses the value for "host.name" 如果已配置。否则，它将使用 return 从 java.net.InetAddress.getCanonicalHostName() 编辑的值。"

这将只是 return 的 IP，以字符串形式，如果没有在客户端机器的 etc/hosts 中定义，我以前使用，否则它 returns 名称配对IP（在我的例子中是 serverHost1）。另外，我也从未设置过 host.name 的值。

如何在 Java 中使用 Kafka 8.2 API 生成消息？

How can I produce messages with Kafka 8.2 API in Java?

java

sockets

maven

apache-kafka