终端输出+log日志输出目录

以下是输出到kafka的效果图（但是没有输出，我这kafka配置有问题，我用的最新的jdk也找不到错误），虽然没有输出，但是kafka控制台显示了相关的接受信息

可能是如下的错误，有知道解决方案的网友感谢告知

代码区

main.go

package main

import(
    "fmt"
    "github.com/astaxie/beego/logs"
    "logagent/kafka"
    "logagent/tailf"
    // "time"
)

func main() {
    /* 加载配置文件logagent.conf信息 */
    filename := "E:/golang/go_pro/logagent.conf"
    err := loadConf("ini",filename)
    if err != nil {
        fmt.Printf("load conf Failed,err:%v\n",err)
        panic("load conf Failed")
        return
    }

    /* 初始化beego/logs的一些功能，设定输出目录 */
    err = initLogger()
    if err != nil {
        fmt.Printf("load logger Failed,err)
        panic("load logger Failed")
        return
    }

    /*先测试将log输出配置正确，输出到logagent.log中*/
    logs.Debug("load conf succ,config:%v",appConfig)

    /*初始化tailf日志组件 */
    //appConfig.collectConf [{E:\golang\go_pro\logs\logagent.log Nginx_log}]
    fmt.Println("appConfig.collectConf",appConfig.collectConf)
    err = tailf.InitTail(appConfig.collectConf,appConfig.chanSize)
    if err != nil {
        logs.Error("init tail Failed,err:%v",err)
        return
    }
    /*先测试将tailf配置正确，输出到logagent.log中*/
    logs.Debug("initialize tailf succ")

    /*初始kafka的工作*/
    err = kafka.InitKafka(appConfig.kafkaAddr)
    if err != nil {
        logs.Error("init tail Failed,err)
        return
    }

    logs.Debug("initialize all succ")

    err = serverRun()
    if err != nil {
        logs.Error("serverRUn Failed,err)
        return
    }

    logs.Info("program exited")
}

config.go

package main


import(
    "fmt"
    "errors"
    "github.com/astaxie/beego/config"
    "logagent/tailf"
)

var (
    appConfig *Config
)

type Config struct {
    logLevel string
    logPath string

    chanSize int
    kafkaAddr string
    collectConf []tailf.CollectConf
}

func loadCollectConf(conf config.Configer) (err error ) {

    var cc tailf.CollectConf
    cc.LogPath = conf.String("collect::log_path")
    if len(cc.LogPath) == 0 {
        err = errors.New("invalid collect::log_path")
        return
    }

    cc.Topic = conf.String("collect::topic")
    if len(cc.LogPath) == 0 {
        err = errors.New("invalid collect::topic")
        return
    }

    appConfig.collectConf = append(appConfig.collectConf,cc)
    return
}

/*
    加载配置文件信息
    [logs]
    log_level=debug
    log_path=E:\golang\go_pro\logs\logagent.log
    [collect]
    log_path=E:\golang\go_pro\logs\logagent.log
    topic=Nginx_log

    chan_size=100
    [kafka]
    server_addr=192.168.21.8:9092
*/
func loadConf(confType,filename string) (err error) {

    conf,err := config.NewConfig(confType,filename)
    if err != nil {
        fmt.Println("new config Failed,err)
        return
    }
    /*定义一个全局变量保存
    var appConfig *Config
    */
    appConfig = &Config{}
    appConfig.logLevel = conf.String("logs::log_level")
    if len(appConfig.logLevel) == 0 {
        appConfig.logLevel = "debug"
    }

    appConfig.logPath = conf.String("logs::log_path")
    if len(appConfig.logPath) == 0 {
        appConfig.logPath = "E:\\golang\\go_pro\\logs\\logagent.log"
    }

    appConfig.chanSize,err = conf.Int("collect::chan_size")
    if err != nil {
        appConfig.chanSize = 100
    }

    appConfig.kafkaAddr = conf.String("kafka::server_addr")
    if len(appConfig.kafkaAddr) == 0 {
        err = fmt.Errorf("invalid kafka addr")
        return
    }

    err = loadCollectConf(conf)
    if err != nil {
        fmt.Printf("load collect conf Failed,err)
        return
    }
    return 
}

log.go

package main


import (
    "encoding/json"
    "fmt"
    "github.com/astaxie/beego/logs"
)

func convertLogLevel(level string) int {

    switch (level) {
        case "debug":
            return logs.LevelDebug
        case "warn":
            return logs.LevelWarn
        case "info":
            return logs.LevelInfo
        case "trace":
            return logs.LevelTrace
    }

    return  logs.LevelDebug
}

/* 初始化beego/logs的一些功能,设定输出目录 */
func initLogger()(err error) {

    config := make(map[string]interface{})
    config["filename"] = appConfig.logPath
    config["level"] = convertLogLevel(appConfig.logLevel)

    configStr,"level":7}
    fmt.Println(string(configStr))
    return
}

tailf.go

package tailf

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

type CollectConf struct {
    LogPath string
    Topic   string
}
/*{E:\golang\go_pro\logs\logagent.log Nginx_log} 每条配置 */
type TailObj struct {
    tail *tail.Tail
    conf CollectConf
}

type TextMsg struct {
    Msg string
    Topic string
}

type TailObjMgr struct {
    tailObjs []*TailObj
    msgChan chan *TextMsg
}

var (
    tailObjMgr* TailObjMgr
)

func GetOneLine()(msg *TextMsg) {
    msg = <- tailObjMgr.msgChan
    return
}
/*初始化Tail组件一些功能*/
func InitTail(conf []CollectConf,}

        tailObjMgr.msgChan <- textMsg
    }
}

kafka.go

package kafka

import(
    "github.com/Shopify/sarama"
    "github.com/astaxie/beego/logs"
)

var (
    client sarama.SyncProducer 
)

/*初始化kafka*/
func InitKafka(addr string) (err error){

    config := sarama.NewConfig()
    config.Producer.requiredAcks = sarama.WaitForAll
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.Return.Successes = true

    client,err)  
        return
    }
    //记录步骤信息
    logs.Debug("init kafka succ")
    return
}

/* 发送到kafak */
func SendToKafka(data,topic string)(err error) {

    msg := &sarama.ProducerMessage{}
    msg.Topic = topic
    msg.Value = sarama.StringEncoder(data)

    pid,err := client.SendMessage(msg)
    if err != nil {
        logs.Error("send message Failed,data,topic)
        return
    }

    logs.Debug("send succ,topic)
    return
}

server.go

package main
import(
    "logagent/tailf"
    "logagent/kafka"
    "github.com/astaxie/beego/logs"
    "time"
    "fmt"
)


func serverRun() (err error){
    for {
        //从chan中取出
        msg := tailf.GetOneLine()
        fmt.Println(msg)

        err = kafka.SendToKafka(msg.Msg,msg.Topic)

        if err != nil {
            logs.Error("send to kafka Failed,err)
            time.Sleep(time.Second)
            continue
        }
    }
    return
}

logagent.conf

[logs]
@H_403_56@log_level=debug
@H_403_56@log_path=E:\golang\go_pro\logs\logagent.log

[collect]
@H_403_56@log_path=E:\golang\go_pro\logs\logagent.log
@H_403_56@topic=Nginx_log
@H_403_56@chan_size=100

[kafka]
@H_403_56@server_addr=192.168.21.8:9092