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>("
运行程序出现如下错误:
- 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