如何在Java中使用Kafka8.2 API生成消息？

Question

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

我正在使用以下客户端代码创建远程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();

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

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

Answer 1

经过大量的挖掘，我决定实现这里找到的示例：Kafka生产者示例。我缩短了代码，没有实现一个分区器类。我用列出的依赖项更新了我的pom，但我仍然有同样的问题。最后，我做了一些配置更改，一切都工作了。

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

172.xx.xx.xxx     serverHost1

同样，X只是面具。然后，我将server.properties文件中的adveredsed.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的值。