我在windows pc上构建了一个小型测试环境,并编写了以下测试kafka的代码(使用org.apache.kafka中的kafka 2.10:0.9.0.1)。
package iii.functiontesting;
import java.text.ParseException;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
/**
* Hello world!
*
*/
public class test4
{
public static void main( String[] args ) throws ParseException
{
Properties producerProps=new Properties();
producerProps.put("bootstrap.servers", "localhost:9092");
producerProps.put("serializer.class",org.apache.kafka.common.serialization.StringSerializer.class.getName());
producerProps.put("key.serializer",org.apache.kafka.common.serialization.StringSerializer.class.getName());
producerProps.put("value.serializer",org.apache.kafka.common.serialization.StringSerializer.class.getName());
producerProps.put("request.required.acks","1");
KafkaProducer<String,String> kafkawriter= new KafkaProducer<String,String>(producerProps);
ProducerRecord<String,String> msg=new ProducerRecord<>("TEST3","ImKey","teststring1");
kafkawriter.send(msg);
}
}
我使用以下命令检查消息是否正确写入队列
d:\work\kafkaenv\kafka_2.10-0.9.0.1\bin\windows>\kafka-console-consumer.bat--缩放器localhost:2181 --topic 测试3——从头开始
然而,我发现Kafka控制台消费者什么也没显示。
我怀疑我的kafka服务器不能正常运行,所以我使用consoleproducer进行测试。
d:\work\kafkaenv\kafka_2.10-0.9.0.1\bin\windows>\kafka-console-producer.bat--代理列表localhost:9092 --topic 测试3
aaaaa级
这一次,我可以看到aaaaa清楚地显示在控制台消费者。我不知道会发生什么。有人能帮我吗?
1条答案
按热度按时间a7qyws3x1#
你也得打电话
KafkaProducer#flush
[或]KafkaProducer#close
方法,然后终止程序。实际上,生产者在将记录发送给代理之前会缓冲这些记录。看到了吗
buffer.memory
以及batch.size
在Kafka生产者配置