Kafka发送消息方法如下:
Properties properties = new Properties();
properties.put(“zookeeper.connect”, “server1:2181”);
properties.put(“serializer.class”,StringEncoder.class.getName());
properties.put(“metadata.broker.list”,”server1:9092″);
ProducerConfig config = new ProducerConfig(properties);
Producer<String, String> producer = new Producer<String, String>(config);
producer.send(new KeyedMessage<Integer, String>(“
mytopic
“, “Hello Kafka”));
运行程序出现如下错误:
Exception in thread "Thread-0" kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
at kafka.producer.Producer.send(Producer.scala:76)
at kafka.javaapi.producer.Producer.send(Producer.scala:33)
atits.kafka.Producer.run(Producer.java:46)
问题原因:
需要改动kafka安装目录下config文件夹下的server.properties中的以下两个属性
zookeeper.connect=localhost:2181改成zookeeper.connect=ip or domainname:2181
以及默认注释掉的
#host.name=localhost改成host.name=ip or domainname
重新启动kafka:
bin/kafka-server-start.sh config/server.properties &
发送消息成功
在kafka安装目录下执行下列语句查看kafka reciever是否收到消息
bin/kafka-console-consumer.sh –zookeeper [ip or domainname]:2181 –topic [topic name] –from-beginning