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官方教程
