实例：实时读取日志并发送kafka

example.go(做main

package main

import (
    "fmt"
    "sync"

    "example.go/kafka"
    "example.go/tailfile"
    "github.com/Shopify/sarama"
    "github.com/sirupsen/logrus"
    "gopkg.in/ini.v1"
)

type Kafka struct {
    Address string `ini:"addr"`
    Topic   string `ini:"topic"`
    Size    int64  `ini:"chan_size"`
}

type Tsdb struct {
    Address    string `ini:"address"`
    Configpath string `ini:"confpath"`
}

type Configure struct { //映射用结构
    Kafka `ini:"kafka"`
    Tsdb  `ini:"tsdb"`
}

func main() {
    logrus.SetLevel(logrus.DebugLevel)
    var configOBJ = new(Configure) //创建一个configure类型的指针并分配空间
    cfg, err := ini.Load("./configure.ini")
    if err != nil {
        logrus.Error("failed to load configure:", err)
        return
    }
    // err = ini.MapTo(configOBJ,cfg)
    // err := ini.MapTo(configOBJ, "./configure.ini")
    err = cfg.MapTo(configOBJ) //映射
    if err != nil {
        logrus.Error("failed to reflect:", err)
        return
    }
    fmt.Printf("%#v\n", configOBJ)

    err = kafka.Kafkainit([]string{configOBJ.Kafka.Address}, configOBJ.Kafka.Size) //类型转换，kafka初始化
    if err != nil {
        logrus.Error("kafka init failed:", err)
        return
    }
    logrus.Debug("kafka init success")

    err = tailfile.Tailinit(configOBJ.Configpath) //tail初始化
    if err != nil {
        logrus.WithFields(logrus.Fields{"filename": configOBJ.Configpath, "err": err}).Error("file loaded failed")
        return
    }
    logrus.WithFields(logrus.Fields{"filename": configOBJ.Tsdb.Configpath}).Debug("file loaded successfully")

    var wg sync.WaitGroup
    wg.Add(2)
    go run(&wg)
    go send_to_kafka(&wg)
    wg.Wait()

}

func run(wg *sync.WaitGroup) { //从tail中接收读取到的信息并发送给kafka的channel
    defer wg.Done()
    msg := &sarama.ProducerMessage{}
    msg.Topic = "test2"
    for {
        line, ok := <-tailfile.Tail_context.Lines
        if len(line.Text) == 0 {
            continue
        }
        if ok {
            logrus.Info("get message")
            msg.Value = sarama.StringEncoder(line.Text)
            fmt.Println(msg.Value)
            kafka.Kafkareceive(msg)
            logrus.Info("send message")
        } else {
            logrus.Error("failed to get message")
            continue
        }
    }
}

func send_to_kafka(wg *sync.WaitGroup) { //给kafka发送数据
    defer wg.Done()
    defer kafka.Kafka_client.Close()
    for {
        partition, offset, err := kafka.Kafka_client.SendMessage(kafka.Kafkaoutput())
        if err != nil {
            logrus.Error("kafka send failed")
        } else {
            logrus.WithFields(logrus.Fields{"partition": partition, "offset": offset}).Info("send message success")
        }
    }
}

//go run ./example.go
//&main.Configure{Kafka:main.Kafka{Address:"127.0.0.1:9092", Topic:"test1", Size:10000}, Tsdb:main.Tsdb{Address:"127.0.0.1", Configpath:"./readme.log"}}
//DEBU[0000] kafka init success
//DEBU[0000] file loaded successfully                      filename=./readme.log
//INFO[0004] get message
//666
//INFO[0004] send message
//INFO[0004] send message success                          offset=1 partition=0
//INFO[0012] get message
//777
//INFO[0012] send message
//INFO[0012] send message success                          offset=2 partition=0
//INFO[0035] get message
//888
//INFO[0035] send message
//INFO[0035] send message success                          offset=3 partition=0
//INFO[0039] get message
//999
//INFO[0039] send message
//INFO[0039] send message success                          offset=4 partition=0

tailread.go

package tailfile

import (
    "fmt"

    "github.com/nxadm/tail"
)

var (
    Tail_context *tail.Tail
    //Context_listen chan *tail.Line
)

func Tailinit(filename string) (err error) {
    config := tail.Config{
        ReOpen:    true, //当出现轮转切割的时候，这两个参数会去追踪新的文件
        Follow:    true,
        Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, //从哪开始
        MustExist: false,                                //允许文件不存在
        Poll:      true,                                 //轮询
    }

    Tail_context, err = tail.TailFile(filename, config)
    if err != nil {
        fmt.Println("error happens:", err)
    }
    return err
}

kafka.go
package kafka

import (
    "github.com/Shopify/sarama"
    "github.com/sirupsen/logrus"
)

var (
    Kafka_client sarama.SyncProducer
    send_chan    chan *sarama.ProducerMessage
)

func Kafkainit(address []string, size int64) (err error) {
    config := sarama.NewConfig()                              //配置
    config.Producer.RequiredAcks = sarama.WaitForAll          //ack应答模式为all
    config.Producer.Partitioner = sarama.NewRandomPartitioner //partition的选择方式为随机
    config.Producer.Return.Successes = true                   //Successes channel 返回值

    send_chan = make(chan *sarama.ProducerMessage, size)

    Kafka_client, err = sarama.NewSyncProducer(address, config) //使用参数创建一个SyncProducer
    //    defer kafka_client.Close()
    if err != nil {
        logrus.Error("broker connection has something wrong happened:", err)
        return err
    }
    return err
}

func Kafkareceive(msg *sarama.ProducerMessage) {
    send_chan <- msg
}

func Kafkaoutput() *sarama.ProducerMessage {
    return <-send_chan
}

configure.ini
[kafka]
addr = "127.0.0.1:9092"
topic = "test1"
chan_size = 10000

[tsdb]
address = 127.0.0.1
confpath = "./readme.log"

bin/kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic test2运行消费者测试成功