如何在主机上使用golang访问docker中安装的kafka

我需要使用golang来访问kafka，所以我在Docker中安装了一个kafka＆zookepper。

1.卡夫卡安装脚本：

# pull images docker pull wurstmeister/zookeeper docker pull wurstmeister/kafka # run kafka & zookepper docker run -d --name zookeeper -p 2181 -t wurstmeister/zookeeper docker run --name kafka -e HOST_IP=localhost -e KAFKA_ADVERTISED_PORT=9092 -e KAFKA_BROKER_ID=1 -e ZK=zk -p 9092:9092 --link zookeeper:zk -t wurstmeister/kafka # enter container docker exec -it ${CONTAINER ID} /bin/bash cd opt/kafka_2.11-0.10.1.1/ # make a tpoic bin/kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic mykafka # start a producer in terminal-1 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic mykafka # start another terminal-2 and start a consumer bin/kafka-console-consumer.sh --zookeeper zookeeper:2181 --topic mykafka --from-beginning 

当我在生产者中input一些消息时，消费者会马上得到它。 所以我认为卡夫卡工作正常

2.现在我需要用golang创build一个消费者来访问kafka。

这里是我的golang演示代码：

 import "github.com/bsm/sarama-cluster" func Consumer(){ // init (custom) config, enable errors and notifications config := cluster.NewConfig() config.Consumer.Return.Errors = true config.Group.Return.Notifications = true // init consumer brokers := []string{"192.168.9.100:9092"} topics := []string{"mykafka"} consumer, err := cluster.NewConsumer(brokers, "my-group-id", topics, config) if err != nil { panic(err) } defer consumer.Close() // trap SIGINT to trigger a shutdown. signals := make(chan os.Signal, 1) signal.Notify(signals, os.Interrupt) // consume messages, watch errors and notifications for { select { case msg, more := <-consumer.Messages(): if more { fmt.Fprintf(os.Stdout, "%s/%d/%d\t%s\t%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value) consumer.MarkOffset(msg, "") // mark message as processed } case err, more := <-consumer.Errors(): if more { log.Printf("Error: %s\n", err.Error()) } case ntf, more := <-consumer.Notifications(): if more { log.Printf("Rebalanced: %+v\n", ntf) } case <-signals: return } } 

}

实际上这个演示代码是从github回购的演示中复制的： sarama-cluster

运行代码时，我得到一个错误：

 kafka: client has run out of available brokers to talk to (Is your cluster reachable?) 

我在启动kafka时使用了端口映射，但是在golang中无法访问它

有没有办法使用curl访问卡夫卡？ 我试过了：

 curl http://192.168.99.10:9092 

和kafka报告错误：

 [2017-08-02 06:39:15,232] WARN Unexpected error from /192.168.99.1; closing connection (org.apache.kafka.common.network.Selector) org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 1195725856 larger than 104857600) at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:95) at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:75) at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:203) at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:167) at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:379) at org.apache.kafka.common.network.Selector.poll(Selector.java:326) at kafka.network.Processor.poll(SocketServer.scala:499) at kafka.network.Processor.run(SocketServer.scala:435) at java.lang.Thread.run(Thread.java:748) 

BTW：

我使用Windows 7

dcoker机器的ip：192.168.99.100

这让我疯狂

有一些build议或解决scheme吗？ 欣赏！！！

Answer 1

不确定，如果可以使用与卡夫卡curl。 但是你可以使用kafka-console-consumer。

 kafka-console-consumer.bat --bootstrap-server 192.168.9.100:9092 --topic mykafka --from-beginning 

Answer 2

我find了原因。 因为卡夫卡的设置不正确

这是server.properties ：

 ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #listeners=PLAINTEXT://:9092 # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #advertised.listeners=PLAINTEXT://your.host.name:9092 

如果listeners器没有设置，kafka将只接收来自java.net.InetAddress.getCanonicalHostName()请求，这意味着localhost

所以我shuld集：

 listeners = PLAINTEXT://0.0.0.0:9092 

这将工作