LogAgent的工作流程：#

1. 读日志 —tailf第三方库

   1
   func main() {
   2
     fileName := "./my.log"
   3
     config := tail.Config{
   4
       ReOpen:    true,                                 // 重新打开
   5
       Follow:    true,                                 // 是否跟随
   6
       Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, // 从文件的哪个地方开始读
   7
       MustExist: false,                                // 文件不存在不报错
   8
       Poll:      true,
   9
     }
   10
     tails, err := tail.TailFile(fileName, config)
   11
     if err != nil {
   12
       fmt.Println("tail file failed, err:", err)
   13
       return
   14
     }
   15
     var (
   16
       line *tail.Line
   17
       ok   bool
   18
     )
   19
     for {
   20
       line, ok = <-tails.Lines
   21
       if !ok {
   22
         fmt.Printf("tail file close reopen, filename:%s\n", tails.Filename)
   23
         time.Sleep(time.Second)
   24
         continue
   25
       }
   26
       fmt.Println("line:", line.Text)
   27
     }
   28
   }
   

2. 往kafka写日志 —sarama第三方库

   1
   // 基于sarama第三⽅库开发的kafka client
   2
   func main() {
   3
     config := sarama.NewConfig()
   4
     // tailf包使⽤
   5
     config.Producer.RequiredAcks = sarama.WaitForAll // 发送完数据需要leader和follow都确认
   6
     config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出⼀个 partition
   7
     config.Producer.Return.Successes = true // 成功交付的消息将在success channel返回
   8
   
   9
   
   10
     // 构造⼀个消息
   11
     msg := &sarama.ProducerMessage{}
   12
     msg.Topic = "web_log"
   13
     msg.Value = sarama.StringEncoder("this is a test log")
   14
   
   15
     // 连接kafka
   16
     client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
   17
     if err != nil {
   18
       fmt.Println("producer closed, err:", err)
   19
       return
   20
     }
   21
     fmt.Println("连接kafka成功！")
   22
     defer client.Close()
   23
     // 发送消息
   24
     pid, offset, err := client.SendMessage(msg)
   25
     fmt.Println("xxx")
   26
     if err != nil {
   27
       fmt.Println("send msg failed, err:", err)
   28
       return
   29
     }
   30
     fmt.Printf("pid:%v offset:%v\n", pid, offset)
   31
     fmt.Println("发送成功！")
   32
   }