5.Go语言项目操作之Kafka日志收集项目实践

Linux & Windows 操作系统: 此处 Ubuntu 20.04 LTS (ip: 10.10.10.225)
JDK 环境 （Java 运行环境-此处不在累述）
kafka 二进制包

# 1.PGP 签名可以使用 PGP 或 GPG 进行验证, 首先下载相关发行版KEYS的asc签名文件。
% gpg --import KEYS 
% gpg --verify 下载文件.asc 下载文件

# 2.验证文件上的哈希,可以使用 GPG 计算哈希：
% gpg --print-md SHA256 下载文件

# Linux Unix 系统（和 macOS）将有一个名为 md5、md5sum 或 shasum 的实用程序。
% md5sum 下载文件

# Windows 7 和更高版本的系统
% certUtil -hashfile 下载文件 

# PS: 哈希摘要算法选择 MD2 MD4 MD5 SHA1 SHA256 SHA384 SHA512。

cd /opt/software/ && mkdir -vp /data/{zookeeper,kafka-logs}
wget https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
tar zxf kafka_2.13-3.0.0.tgz

# 1.针对 zookeeper.properties 配置修改
$ grep -v -E '^#' /opt/software/kafka_2.13-3.0.0/config/zookeeper.properties
dataDir=/data/zookeeper  # 关键点1.存储快照的目录
clientPort=2181          # 关键点2.zookeeper 服务端监听端口
maxClientCnxns=0
admin.enableServer=false

# 2.1.针对 server.properties 配置
$ grep -v -E '^#'  /opt/software/kafka_2.13-3.0.0/config/server.properties
broker.id=0                  # 关键点1.服务器节点ID
listeners=PLAINTEXT://:9092  # 关键点2.kafka 服务端监听端口
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.dirs=/data/kafka-logs   # 关键点3.日志存储路径
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
num.partitions=1            # 分区数量
num.recovery.threads.per.data.dir=1
zookeeper.connect=10.10.107.225:2181   # 关键点4.zookeeper 服务端地址，如有多个请使用,分隔。
zookeeper.connection.timeout.ms=18000  # 关键点5. 连接 zookeeper 超时时间。

# 首先启动
/opt/software/kafka_2.13-3.0.0# ./bin/zookeeper-server-start.sh ./config/zookeeper.properties 2>&1 > /tmp/zookeeper.log &

# 再次执行
/opt/software/kafka_2.13-3.0.0# ./bin/kafka-server-start.sh ./config/server.properties  2>&1 > /tmp/kafka.log &
  # [2022-01-18 21:55:52,024] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
  # [2022-01-18 21:55:52,297] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
  # [2022-01-18 21:55:52,364] INFO Registered signal handlers for TERM, INT, HUP (org.apache.kafka.common.utils.LoggingSignalHandler)
  # [2022-01-18 21:55:52,366] INFO starting (kafka.server.KafkaServer)
  # [2022-01-18 21:55:52,367] INFO Connecting to zookeeper on 10.10.107.225:2181 (kafka.server.KafkaServer)
  # [2022-01-18 21:55:52,378] INFO [ZooKeeperClient Kafka server] Initializing a new session to 10.10.107.225:2181. (kafka.zookeeper.ZooKeeperClient)
  # [2022-01-18 21:55:52,381] INFO Client environment:zookeeper.version=3.6.3--6401e4ad2087061bc6b9f80dec2d69f2e3c8660a, built on 04/08/2021 16:35 GMT (org.apache.zookeeper.ZooKeeper)

➜  weiyigeek.top $ go get github.com/hpcloud/tail 
go: downloading github.com/hpcloud/tail v1.0.0
go: downloading gopkg.in/fsnotify.v1 v1.4.7

package main

import (
  "fmt"
  "time"
  "github.com/hpcloud/tail"
)

// tail 的具体用法示例
func main() {
  // 1.日志文件路径及其tail包Config设置
  fileName := "./access.log"
  config := tail.Config{
    ReOpen:    true,                                 // 是否重新打开
    Follow:    true,                                 // 是否进行跟随
    Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, // 是否从文件的首行开始读取
    MustExist: false,                                // 文件是否必须存在
    Poll:      true,                                 // 是否轮询文件更改而不是使用inotify通知

  }

  // 2.持续跟随文件读取
  tailObj, err := tail.TailFile(fileName, config)
  if err != nil {
    fmt.Printf("Tail file failed, Err:%v", err)
    return
  }

  // 3.读取局部变量定义
  var (
    msg    *tail.Line // 拿取到的每一行
    status bool       // 拿取状态记录
  )

  // 4.跟踪读取文件内容
  for {
    msg, status = <-tailObj.Lines // 日志读取
    if !status {
      fmt.Printf("Tail file Close Reopen filename: %s\n", tailObj.Filename)
      time.Sleep(time.Second)
      continue
    }

    // 带中文输出的两种方式。
    // # 方式1.利用Ascii码转，中文在Go中是UTF8编码占用三个字节。
    // for _, ch := range msg.Text {
    // 	fmt.Printf("%s", string(ch))
    // }

    // # 方式2.利用string函数
    fmt.Println(string(msg.Text))
  }
}

➜  01tail_demo go build .                        
➜  01tail_demo ./01tail_demo      
2021/12/28 03:42:17 Seeked ./access.log - &{Offset:0 Whence:2}
2021/12/28 03:43:06 Re-opening moved/deleted file ./access.log ...
2021/12/28 03:43:06 Successfully reopened ./access.log
Hello World! Tail File Demo!
测试文件: access.log
测试输入中文字符。
Insert into line demo!
My Name is WeiyiGeek!

➜  weiyigeek.top $ 
go get github.com/Shopify/sarama  # 当前 sarama 版本为 v1.30.1。
go: downloading github.com/Shopify/sarama v1.30.1  
.....
go: downloading github.com/pierrec/lz4 v1.0.1
go get: added github.com/Shopify/sarama v1.30.1
.....
go get: added gopkg.in/jcmturner/rpc.v1 v1.1.0

package main

import (
  "fmt"
  "log"
  "os"
  "os/signal"
  "sync"
  "time"

  "github.com/Shopify/sarama"
)

// Go语言中使用第三方库sarama连接kafka服务进行生产，生产者示例。
var (
  wg                          sync.WaitGroup
  enqueued, successes, errors int // 统计参数
)

// 基于sarama第三方库开发的kafka client示例
func main() {
  // 1.NewConfig 返回一个具有正常默认值的新配置实例。
  config := sarama.NewConfig()
  config.Producer.Partitioner = sarama.NewRandomPartitioner // 随机新选出一个partition
  config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
  config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回
  config.Producer.Partitioner = sarama.NewHashPartitioner   // 对Key进行Hash，同样的Key每次都落到一个分区，这样消息是有序的

  // 2.连接到 kafka 服务端
  // client, err := sarama.NewSyncProducer([]string{"10.10.107.225:9092"}, config)
  client, err := sarama.NewClient([]string{"10.10.107.225:9092"}, config)
  if err != nil {
    fmt.Println("producer closed, err:", err)
    return
  }
  defer client.Close()

  // 4.使用给定的客户端创建新的生产者，关闭此生产者时仍然需要在基础客户端上调用Close（）。
  producer, err := sarama.NewAsyncProducerFromClient(client)
  if err != nil {
    panic(err)
  }

  // 5.监听 SIGINT 信号触发程序停止(值得借鉴学习)
  signals := make(chan os.Signal, 1)
  signal.Notify(signals, os.Interrupt)

  wg.Add(1)
  // 6.启动一个 groutines 来计算成功次数
  go func() {
    defer wg.Done()
    for range producer.Successes() {
      successes++
    }
  }()

  wg.Add(1)
  // 7.启动一个 groutines 来计算失败次数
  go func() {
    defer wg.Done()
    for err := range producer.Errors() {
      log.Println(err)
      errors++
    }
  }()

ProducerLoop:
  for {
    // 8.构造一个消息生产者
    // msg := &sarama.ProducerMessage{Topic: "my_topic", Value: sarama.StringEncoder("This is a sarama test log！By WeiyiGeek!")}
    msg := &sarama.ProducerMessage{}
    msg.Topic = "my_topic"
    msg.Key = sarama.StringEncoder("sarama")
    message := fmt.Sprintf("This is a sarama test log！By WeiyiGeek! UnixTime:%v", time.Now().Unix())
    msg.Value = sarama.StringEncoder(message)

    fmt.Println(message)
    // 9.利用通道的形式将信息异步发送给kafka
    select {
    case producer.Input() <- msg:
      enqueued++
      time.Sleep(time.Second * 2)
      if enqueued >= 5 {
        producer.AsyncClose() // 队列数超过5时则停止生产数据。
        break ProducerLoop
      }
    case <-signals:
      producer.AsyncClose() // Trigger a shutdown of the producer.
      break ProducerLoop    // 按Ctrl+c则退出循环
    }
  }
  // 11.等待goroutine任务结束。
  wg.Wait()
  // 10.输出表示发生成功
  log.Printf("Successfully, enqueued: %v ,produced: %d; errors: %d\n", enqueued, successes, errors)
}

package main

import (
  "fmt"
  "sync"

  "github.com/Shopify/sarama"
)

// Go语言中使用sarama第三方库连接kafka服务进行消费，消费者示例。
func main() {

  // 1.连接到 kafka 服务端
  consumer, err := sarama.NewConsumer([]string{"10.10.107.225:9092"}, nil)
  if err != nil {
    fmt.Printf("fail to start consumer, err:%v\n", err)
    return
  }

  // 2.根据topic取到所有的分区
  partitionList, err := consumer.Partitions("my_topic")
  if err != nil {
    fmt.Printf("fail to get list of partition:err%v\n", err)
    return
  }
  fmt.Println("分区信息:", partitionList)

  // 3.遍历所有的分区
  var wg sync.WaitGroup
  wg.Add(1)
  for partition := range partitionList {
    // 4.针对每个分区创建一个对应的分区消费者
    pc, err := consumer.ConsumePartition("my_topic", int32(partition), sarama.OffsetNewest)
    if err != nil {
      fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
      return
    }
    defer pc.AsyncClose()

    // 5.异步从每个分区消费信息
    go func(sarama.PartitionConsumer) {
      defer wg.Done()
      for msg := range pc.Messages() {
        fmt.Printf("Topic: %v ,Partition:%d ,Offset:%d ,Key: %v ,Value: %v \n", msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
      }
    }(pc)
  }
  wg.Wait()
}

# 生产者
➜ 02sarama_producer go build producer.go
➜ 02sarama_producer ./producer
This is a sarama test log！By WeiyiGeek! UnixTime:1640635641
This is a sarama test log！By WeiyiGeek! UnixTime:1640635643
This is a sarama test log！By WeiyiGeek! UnixTime:1640635645
This is a sarama test log！By WeiyiGeek! UnixTime:1640635647
This is a sarama test log！By WeiyiGeek! UnixTime:1640635649
2021/12/28 04:07:31 Successfully, enqueued: 5 ,produced: 5; errors: 0

# 消费者
➜  Day11 go build consumer.go
➜  Day11 ./consumer 
分区信息: [0]
Topic: my_topic ,Partition:0 ,Offset:379686590 ,Key: sarama ,Value: This is a sarama test log！By WeiyiGeek! UnixTime:1640635641 
Topic: my_topic ,Partition:0 ,Offset:379686591 ,Key: sarama ,Value: This is a sarama test log！By WeiyiGeek! UnixTime:1640635643 
Topic: my_topic ,Partition:0 ,Offset:379686592 ,Key: sarama ,Value: This is a sarama test log！By WeiyiGeek! UnixTime:1640635645 
Topic: my_topic ,Partition:0 ,Offset:379686593 ,Key: sarama ,Value: This is a sarama test log！By WeiyiGeek! UnixTime:1640635647 
Topic: my_topic ,Partition:0 ,Offset:379686594 ,Key: sarama ,Value: This is a sarama test log！By WeiyiGeek! UnixTime:1640635649

package main

import (
  "context"
  "fmt"
  "github.com/Shopify/sarama"
  "os"
  "os/signal"
  "sync"
)
type consumerGroupHandler struct{
  name string
}

func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
  for msg := range claim.Messages() {
    fmt.Printf("%s Message topic:%q partition:%d offset:%d  value:%s\n",h.name, msg.Topic, msg.Partition, msg.Offset, string(msg.Value))
    // 手动确认消息
    sess.MarkMessage(msg, "")
  }
  return nil
}

func handleErrors(group *sarama.ConsumerGroup,wg  *sync.WaitGroup ){
  wg.Done()
  for err := range (*group).Errors() {
    fmt.Println("ERROR", err)
  }
}

func consume(group *sarama.ConsumerGroup,wg  *sync.WaitGroup, name string) {
  fmt.Println(name + "start")
  wg.Done()
  ctx := context.Background()
  for {
    topics := []string{"my_topic"}
    handler := consumerGroupHandler{name: name}
    err := (*group).Consume(ctx, topics, handler)
    if err != nil {
      panic(err)
    }
  }
}

func main(){
  var wg sync.WaitGroup
  config := sarama.NewConfig()
  config.Consumer.Return.Errors = false
  config.Version = sarama.V0_10_2_0
  client,err := sarama.NewClient([]string{"localhost:9192","localhost:9292","localhost:9392"}, config)
  defer client.Close()
  if err != nil {
    panic(err)
  }
  group1, err := sarama.NewConsumerGroupFromClient("c1", client)
  if err != nil {
    panic(err)
  }
  group2, err := sarama.NewConsumerGroupFromClient("c2", client)
  if err != nil {
    panic(err)
  }
  group3, err := sarama.NewConsumerGroupFromClient("c3", client)
  if err != nil {
    panic(err)
  }
  defer group1.Close()
  defer group2.Close()
  defer group3.Close()
  wg.Add(3)
  go consume(&group1,&wg,"c1")
  go consume(&group2,&wg,"c2")
  go consume(&group3,&wg,"c3")
  wg.Wait()
  signals := make(chan os.Signal, 1)
  signal.Notify(signals, os.Interrupt)
  select {
  case <-signals:
  }
}

在Windows平台编译时会提示类似如下错误，解决办法:在Windows平台请使用v1.19版本的sarama。
# github.com/DataDog/zstd
exec: "gcc":executable file not found in %PATH%

# 错误信息
send msg failed, err: dial tcp: lookup es01: Temporary failure in name resolution

# 解决办法
tee -a /etc/hosts <<'EOF'
10.10.107.225 es01
EOF