Golang zap框架应用(二)——将日志信息输出到多个输出流中

Bertha 。 2024-04-19 09:50 91阅读 0赞

目标功能：

1.将error以下级别添加到info.log文件中

2.将error及以上添加到error.log文件中

3.将日志数据写入到kafka中(kafka客户端使用sarama)

下载：

go get -u go.uber.org/zap

go get gopkg.in/Shopify/sarama.v1

var(
    	producerTopic = "klog"
    )
    
    
    func main() {
    	logger,err  := getLoggerWithKafka("D:/log/kinfo.log","D:/log/kerror.log")
    	if err != nil {
    		log.Fatal(err)
    	}
    
    
    	for i := 0 ; i < 10 ;i++ {
    		logger.Debug("i am debug",zap.String("key","debug"))
    		logger.Error("i am error",zap.String("key","error"))
    		logger.Info("i am info",zap.String("key","info"))
    		time.Sleep(time.Second)
    	}
    }
    
    
    func getLoggerWithKafka(infoPath,errorPath string)  (*zap.Logger,error) {
    	kp,err := newKafkaProducer(producerTopic,"localhost:9092")
    	if err != nil {
    		log.Fatal(err)
    	}
    	kws := zapcore.AddSync(kp)
    
    
    	highPriority := zap.LevelEnablerFunc(func(lev zapcore.Level) bool{
    		return lev >= zap.ErrorLevel
    	})
    
    	lowPriority := zap.LevelEnablerFunc(func(lev zapcore.Level) bool {
    		return lev < zap.ErrorLevel && lev > zap.DebugLevel
    	})
    
    	kafkaPriority := zap.LevelEnablerFunc(func(lev zapcore.Level) bool{
    		return lev >= zap.DebugLevel
    	})
    
    
    	prodEncoder := zap.NewProductionEncoderConfig()
    	prodEncoder.EncodeTime = zapcore.ISO8601TimeEncoder
    
    
    	lowWriteSyncer,lowClose,err :=  zap.Open(infoPath)
    	if err != nil {
    		lowClose()
    		return nil,err
    	}
    
    
    	highWriteSyncer,highClose,err :=  zap.Open(errorPath)
    	if err != nil {
    		highClose()
    		return nil,err
    	}
    
    	highCore := zapcore.NewCore(zapcore.NewJSONEncoder(prodEncoder),highWriteSyncer,highPriority)
    	lowCore := zapcore.NewCore(zapcore.NewJSONEncoder(prodEncoder),lowWriteSyncer,lowPriority)
    	kafkaCore := zapcore.NewCore(zapcore.NewJSONEncoder(prodEncoder),kws,kafkaPriority)
    
    	return  zap.New(zapcore.NewTee(highCore,lowCore,kafkaCore),zap.AddCaller()),nil
    }
    
    
    func newKafkaProducer(topic,addr string) (*kafkaProducer,error) {
    	config := sarama.NewConfig()
    	config.Producer.RequiredAcks = sarama.WaitForLocal
    	config.Producer.Return.Successes = true
    	syncProducer ,err := sarama.NewSyncProducer([]string{addr},config)
    	if err != nil {
    		return nil,err
    	}
    
    	return &kafkaProducer{
    		topic:    topic,
    		syncProducer: syncProducer,
    	},nil
    }
    
    
    type kafkaProducer struct {
    	topic string
    	syncProducer sarama.SyncProducer
    }
    
    func (kp *kafkaProducer) Write(p []byte) (n int, err error)  {
    	_,_, err = kp.syncProducer.SendMessage(&sarama.ProducerMessage{
    		Topic: kp.topic,
    		Value: sarama.ByteEncoder(p),
    	})
    	if err != nil {
    		log.Println(err)
    		return 0,err
    	}
    	//fmt.Println("send msgs:",string(p))
    	return len(p),nil
    }
    
    func (kp *kafkaProducer) Close(p []byte)   {
    	kp.syncProducer.Close()
    }