Go语言与Docker操作Kafka

Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据，具有高性能、持久化、多副本备份、横向扩展等特点。

kafka.jpeg

本文介绍了如何使用Go语言发送和接收kafka消息。操作系统为Ubuntu 18版本。使用Go语言中连接kafka使用第三方库:github.com/Shopify/sarama。本教程使用最新版kafka_2.13-2.6.0（Scala为2.13，Kafka为2.6）

首先下载sarama库

go get github.com/Shopify/sarama

Docker中安装运行kafka

1、拉取zookeeper和kafka镜像文件

docker pull wurstmeister/zookeeper
docker pull wurstmeister/kafka

2、先运行zookeeper，再运行kafka

 ~$ docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

~$ docker run -d --name kafka --publish 9092:9092 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 --env KAFKA_ADVERTISED_HOST_NAME=127.0.0.1 --env KAFKA_ADVERTISED_PORT=9092 --volume /etc/localtime:/etc/localtime wurstmeister/kafka:latest

3、docker查看运行状态

~$ docker ps
CONTAINER ID        IMAGE                       COMMAND                  CREATED             STATUS              PORTS                                                NAMES
1375a7b86b0c        wurstmeister/kafka:latest   "start-kafka.sh"         23 seconds ago      Up 23 seconds       0.0.0.0:9092->9092/tcp                               kafka
3b66998752b5        wurstmeister/zookeeper      "/bin/sh -c '/usr/sb…"   33 seconds ago      Up 33 seconds       22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp   zookeeper

4、使用Go语言创建Topic，并发送数据至kafka

package main

import (
    "github.com/Shopify/sarama"
    "log"
    "time"
    "fmt"
)

func main() {
    config := sarama.NewConfig()
    // request.timeout.ms
    config.Producer.Timeout = time.Second * 5
    // message.max.bytes
    config.Producer.MaxMessageBytes = 1024 * 1024
    // request.required.acks
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Return.Successes=true
    config.Version = sarama.V0_11_0_1

    if err := config.Validate(); err != nil {
        panic(fmt.Errorf("invalid configuration, error: %v", err))
    }

    producer, err := sarama.NewSyncProducer([]string{"172.17.0.1:9092"}, config)
    if err != nil {
        log.Fatalln(err)
    }
    defer func() {
        if err := producer.Close(); err != nil {
            log.Fatalln(err)
        }
    }()

    for {
        msg := &sarama.ProducerMessage{
            Topic: "topic-C",
            Value: sarama.StringEncoder("this is a testing message from sarama!!!"),
        }
        partition, offset, err := producer.SendMessage(msg)
        if err != nil {
            log.Printf("FAILED to send message: %s\n", err)
        } else {
            log.Printf("> message sent to partition %d at offset %d\n", partition, offset)
        }
        time.Sleep(1 * time.Second)
    }
}

输出成功为：

2020/12/01 18:08:02 > message sent to partition 0 at offset 5581
2020/12/01 18:08:03 > message sent to partition 0 at offset 5582
2020/12/01 18:08:04 > message sent to partition 0 at offset 5583
2020/12/01 18:08:05 > message sent to partition 0 at offset 5584

5、查看对应的topic的描述信息

~$ docker exec kafka kafka-topics.sh --describe --topic topic-C --zookeeper zookeeper:2181
Topic: topic-C  PartitionCount: 1   ReplicationFactor: 1    Configs: 
    Topic: topic-C  Partition: 0    Leader: 1001    Replicas: 1001  Isr: 1001

6、查看当前kafka里有哪些Topic

~$ docker exec kafka kafka-topics.sh --list --zookeeper zookeeper:2181
topic-C

7、查看Topic里面的数据

~$ docker exec kafka kafka-console-consumer.sh --topic topic-C  --from-beginning --bootstrap-server localhost:9092
this is a testing message from sarama!!!
this is a testing message from sarama!!!
this is a testing message from sarama!!!
this is a testing message from sarama!!!
this is a testing message from sarama!!!
this is a testing message from sarama!!!
this is a testing message from sarama!!!

看到数据已经存在于Kafka的Topic里面，发送成功！

Docker操作Kafka其他命令

1、创建一个新的topic

~$ docker exec kafka kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
Created topic quickstart-events.

2、查看当前docker中kafka版本号

~$ docker exec kafka find / -name \*kafka_\* | head -1 | grep -o '\kafka[^\n]*'
kafka_2.13-2.6.0

3、向一个Topic手动写入消息数据

$ docker exec kafka kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
hello world
I Love You!

其他命令可查看Kafka官方教程