我有一个用 Java 编写的 Kafka 生产者。尽管它基本上是示例代码的剪切和过去,但它似乎无法正常工作。我预计向我的集群输出 10 条消息。相反,我得到了消息成功输出,但实际上没有任何内容发送到我的集群。我不确定从哪里开始排除故障。
import java.util.Properties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class SimpleProducer {
public static void main(String[] args) throws Exception{
String topicName = "test_topic";
Properties props = new Properties();
props.put("bootstrap.servers", "skynet.local:6667");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);
for(int i = 0; i < 10; i++)
producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(i), Integer.toString(i)));
System.out.println("Message sent successfully");
producer.close();
}
}
最佳答案
由于某些环境尚未清除,所以我会根据您的 Kafka 服务器已在端口 6667 上工作来尝试回答您的问题。
您的代码可能需要调整两个地方(有人可以帮助我改进它):
props.put("linger.ms", 1); // set to 0 let Producer can send message immediately
在这里,将 Producer.close();
从 for
循环中删除:
Producer<String, String> producer = new KafkaProducer<String, String>(props, new StringSerializer(), new StringSerializer());
for(int i = 0; i < 10; i++) {
Future<RecordMetadata> f = producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(i), Integer.toString(i)));
System.out.println(f.get()); // don't do that in your Production, here just for debugging purpose.
}
producer.close();
还有一件事,您可以在测试之前运行 kafka-console-consumer.sh
和 kafka-console- Producer.sh
以确认您的 Kafka 服务器和 SimpleProducer已经在工作了。 Kafka 0.10.x 配置参数位于 Kafka Producer Configuration Parameters
关于java - Kafka Producer 似乎无法正常工作,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43459962/