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

Question

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

1. kafka安装脚本如下：

# 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 

当我在producer中输入一些消息时，消费者会立即得到它。所以我假设卡夫卡工作得很好

2.现在我需要创建一个带有golang的消费者来访问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访问kafka？我试过了：

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)

顺便说一句：

我使用的是windows 7

dcoker机器ip :192.168.99.100

它把我逼疯了

有什么建议或解决方案吗？感谢！

Answer 1

如果你想创建一个消费者来收听来自Kafka的话题，让我们试一试。我使用了教程中的confluent-kafka-go：https://github.com/confluentinc/confluent-kafka-go

这是main.go文件上的代码：

import (
    "fmt"
    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)

func main() {

    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost",
        "group.id":          "myGroup",
        "auto.offset.reset": "earliest",
    })

    if err != nil {
        panic(err)
    }

    c.SubscribeTopics([]string{"myTopic", "^aRegex.*[Tt]opic"}, nil)

    for {
        msg, err := c.ReadMessage(-1)
        if err == nil {
            fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
        } else {
            // The client will automatically try to recover from all errors.
            fmt.Printf("Consumer error: %v (%v)\n", err, msg)
        }
    }

    c.Close()
}

如果您使用docker进行构建:请遵循此注释添加合适的包

对于基于Debian和Ubuntu的发行版，请从标准存储库或使用Confluent的Deb存储库安装librdkafka-dev。

对于基于Redhat的发行版，使用Confluent的YUM存储库安装librdkafka-devel。

对于MacOS X，安装来自Homebrew的librdkafka。您可能还需要brew安装pkg-config (如果尚未安装)。brew安装librdkafka pkg-config。

对于高山: apk add librdkafka-dev pkgconf confluent-kafka-go在Windows上不受支持。

请记住安装社区版本，因为它不能安装最大版本1.1.0的

(不使用阿尔卑斯社区版本)

祝好运!

Answer 2

不确定，是否可以将curl与kafka一起使用。但是您可以使用kafka-console-consumer。

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

Answer 3

我已经找到原因了。因为kafka的设置不正确

这是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

所以我应该设置：

listeners = PLAINTEXT://0.0.0.0:9092

这将会起作用