异步处理日志切换后文件关闭工作

Author: Chen Ming

file.go

@@ -9,9 +9,26 @@ import (
 // filepath 文件默认地址
 // nextTheTime下一个整点时间
 // 返回整点时间日志文件地址
-func GenLogFilepath(filepath string, nextTheTime time.Time) string {
+func GenLogFilepath(filepath string, nextTheTime time.Time, tk time.Duration) string {
 	fileSlice := strings.Split(filepath, ".")
-	return strings.Join([]string{fileSlice[0], nextTheTime.Format("2006-01-02 15:04:05"), ".log"}, "")
+	var formatStyle string
+	switch tk {
+	case time.Nanosecond:
+		formatStyle = "20060102150405000000000"
+	case time.Second:
+		formatStyle = "20060102150405"
+	case time.Minute:
+		formatStyle = "200601021504"
+	case time.Hour:
+		formatStyle = "2006010215"
+	case TheDay:
+		formatStyle = "20060102"
+	case TheMonth:
+		formatStyle = "200601"
+	case TheYear:
+		formatStyle = "2006"
+	default:
+		formatStyle = "2006010215"
+	}
+	return strings.Join([]string{fileSlice[0], nextTheTime.Format(formatStyle), ".log"}, "")
 }
-
-

file_test.go

@@ -9,7 +9,7 @@ func TestGenLogFilepath(t *testing.T) {
 	result := "my2021-10-09 00:00:00.log"
 	file := "my.log"
 	dur := time.Date(2021, 10, 9, 0, 0, 0, 0, time.Local)
-	log_name := GenLogFilepath(file, dur)
+	log_name := GenLogFilepath(file, dur,time.Minute)
 	if log_name != result {
 		t.Errorf("生成log_name不正确 result:%s log_naem:%s", result, log_name)
 		return

logger.go

@@ -7,20 +7,8 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
-// zapper 默认的logger的配置
-const (
-	defaultLevel    = zap.DebugLevel
-	defaultProduct  = "zapper"
-	defaultModule   = "zapper"
-	defaultFilePath = "./zapper.log"
-)
-
 // defaultLogger 默认的日志对象
-var defaultLogger *zap.Logger = NewLogger(defaultLevel, defaultProduct, defaultModule, defaultFilePath)
-
-func init()  {
-	defaultLogger.Info("zapper init finished", zap.Bool("success", true),zap.Int("int", 100),zap.String("string", "val string"))
-}
+var defaultLogger *zap.Logger
 
 // 初始化函数
 // level 日志收集的最低级别
@@ -34,14 +22,14 @@ func Zapper(level zapcore.Level, product, module string, outputPath string) {
 }
 
 // Sync 在服务停止前调用,日志罗盘goroutine生命周期管理的收尾工作
+// 通知writer 及时停止；
 // defer zapper.Sync()
 func Sync() {
-	defaultLogger.Sync() 
-	// TODO: 通知writer 及时停止；
+	defaultLogger.Sync()
 	defaultWriter.Close()
 }
 
-// SetDefaultLogger 
+// SetDefaultLogger
 // logger 用于替换zapper中默认的logger
 func SetDefaultLogger(logger *zap.Logger) { defaultLogger = logger }
 

middleware.go

@@ -11,6 +11,10 @@ func LogMiddleware(h http.Handler) http.Handler {
 
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 
+		// TODO:
+		// req body
+		// resp body
+		// latency 响应时长
 		newLogger := wrapReqFields(defaultLogger, r)
 
 		r.WithContext(context.WithValue(r.Context(), "logger", newLogger))
@@ -27,6 +31,7 @@ func LogMiddleware(h http.Handler) http.Handler {
 func wrapReqFields(logger *zap.Logger, r *http.Request) *zap.Logger {
 	urlQuery := r.URL.Query()
 	return logger.With(
+		// TODO: 找到字段所在报文位置
 		zap.String("logid", urlQuery.Get("")),
 		zap.String("caller_ip", r.Header.Get("")),
 		zap.String("host_ip", r.Host),

middleware_test.go

@@ -6,13 +6,15 @@ import (
 	"log"
 	"net/http"
 	"testing"
+
+	"go.uber.org/zap"
 )
 
 func TestLogMiddleware(t *testing.T) {
-	
 
 	fooHandler := func(w http.ResponseWriter, r *http.Request) {
-		w.Write([]byte("I am ok"))
+		Error("LogMiddleware", zap.String("net/http middleware", "net/http的handler打印"))
+		w.Write([]byte("I'm OK"))
 	}
 
 	http.Handle("/foo", LogMiddleware(http.HandlerFunc(fooHandler)))

write.go

@@ -23,13 +23,14 @@ func SetDefaultWriter(w *Writer) { defaultWriter = w }
 // tk 为整点间隔单位
 // tkSStop 为日志切割停止信号
 type Writer struct {
-	buf         chan string
-	signal      chan struct{}
-	f           *os.File
-	nextTheTime time.Time
-	filepath    string
-	tk          time.Duration
-	tkStop      chan struct{}
+	buf          chan string   // 日志数据缓冲
+	rotateSignal chan struct{} // 日志切割信号
+	f            *os.File      // 当前落盘日志文件指针
+	nextTheTime  time.Time     // 当前日志所在整点时间
+	filepath     string        // 默认日志文件地址
+	tk           time.Duration // 日志切割间隔整点时间
+	tkStop       chan struct{} // 日志切割停止信号
+	syncFileBuf  chan *os.File // 需要异步关闭的文件指针
 }
 
 // Clone 复制一个新的Writer
@@ -64,19 +65,17 @@ func (f WriterOptionFunc) Apply(w *Writer) {
 // file 文件名
 // tk 日志切割的间隔整点时间单位
 // cacheMax 缓存的最大值
-// TODO: Writer 的配置项做成Options方式处理
+// Writer 的配置项做成Options方式处理
 func NewWriter(file string, tk time.Duration, cacheMax int64) (w *Writer) {
 	// 路径转换（相对路径转绝对路径）
-	if !filepath.IsAbs(file) {
-		var err error
-		file, err = filepath.Abs(file)
-		if err != nil {
-			panic(err)
-		}
+	var err error
+	file, err = filepath.Abs(file) //绝对路径会直接返回绝对路径
+	if err != nil {
+		panic(err)
 	}
 
 	firstTheTime := NextTheTime(time.Now(), tk)
-	firstFile := GenLogFilepath(file, firstTheTime)
+	firstFile := GenLogFilepath(file, firstTheTime, tk)
 	f, err := os.OpenFile(firstFile, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0666)
 	if err != nil {
 		panic(err)
@@ -87,17 +86,19 @@ func NewWriter(file string, tk time.Duration, cacheMax int64) (w *Writer) {
 	}
 
 	w = &Writer{
-		buf:         make(chan string, cacheMax),
-		signal:      make(chan struct{}),
-		f:           f,
-		nextTheTime: firstTheTime,
-		filepath:    file,
-		tk:          tk,
-		tkStop:      make(chan struct{}),
+		buf:          make(chan string, cacheMax),
+		rotateSignal: make(chan struct{}),
+		f:            f,
+		nextTheTime:  firstTheTime,
+		filepath:     file,
+		tk:           tk,
+		tkStop:       make(chan struct{}),
+		syncFileBuf:  make(chan *os.File, 128),
 	}
 
 	go w.run()
 	go w.rotate()
+	go w.syncSync()
 
 	return
 }
@@ -106,7 +107,6 @@ func NewWriter(file string, tk time.Duration, cacheMax int64) (w *Writer) {
 // io.Writer interface的实现
 func (w *Writer) Write(p []byte) (n int, err error) {
 	w.buf <- string(p)
-	// TODO: whether n,err have been used by zap
 	return
 }
 
@@ -125,61 +125,67 @@ func (w *Writer) run() {
 		case b, ok := <-w.buf:
 			if !ok {
 				w.tkStop <- struct{}{}
-				w.f.Sync()
+				w.f.Sync() // TODO: 检查logger.Sync()
 				w.f.Close()
 				return
 			}
-			// TODO: check json repeated key
+			// check json repeated key
 			fmt.Println("准备写入文件的日志内容:", b)
 			bm := make(map[string]interface{}, 0)
 			err := json.Unmarshal([]byte(b), &bm)
 			if err != nil {
 				fmt.Println(err)
 			}
 
-			fmt.Println("准备写入文件的日志内容,json->map:", bm) //TODO: delete
 			bb, err := json.Marshal(bm)
 			if err != nil {
 				fmt.Println(err)
 			}
-			fmt.Println("当前日志写入的文件名", w.f.Name()) //TODO: delete
 			// w.f.WriteString(b)
 			w.f.WriteString(string(bb) + "\n")
-		case <-w.signal:
-			w.f.Sync()
-			w.f.Close()
+		case <-w.rotateSignal:
+			// 异步化处理文件同步、关闭
+			w.syncFileBuf <- w.f
 			// replace os.File
-			fmt.Println("收到替换信号,当前日志文件:", w.f.Name())
-			fmt.Println("收到替换信号,w.nextTheTime:", w.nextTheTime.Format(time.RFC3339))
-			newFile := GenLogFilepath(w.filepath, w.nextTheTime)
-			fmt.Println("收到替换信号,newFile生成的文件名:", newFile)
+			newFile := GenLogFilepath(w.filepath, w.nextTheTime, w.tk)
 			var err error
 			w.f, err = os.OpenFile(newFile, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0666)
 			if err != nil {
 				fmt.Println(err)
 			}
-			fmt.Println("收到替换信号,已替换替换的日志文件名:", w.f.Name())
 		}
 	}
 }
 
-// rotate 
+// rotate
 // 定时发送日志切割信号
 // 监听切割终止信号
 func (w *Writer) rotate() {
 	for {
 		t := time.NewTimer(w.nextTheTime.Sub(time.Now()))
 		select {
 		case <-w.tkStop:
+			close(w.syncFileBuf) // 通知异步任务关闭
 			return
 		case <-t.C:
 			// change nextTheTime
-			fmt.Println("发送替换日志文件信号时间,NextTheTime前:", w.nextTheTime.Format(time.RFC3339)) // TODO: delete
 			w.nextTheTime = NextTheTime(w.nextTheTime, w.tk)
-			fmt.Println("发送替换日志文件信号时间,NextTheTime后:", w.nextTheTime.Format(time.RFC3339)) // TODO: delete
 			// it's about time to repalce os.File.
-			w.signal <- struct{}{}
+			w.rotateSignal <- struct{}{}
 		}
 
 	}
 }
+
+func (w *Writer) syncSync() {
+	for {
+		select {
+		case f, ok := <-w.syncFileBuf:
+			if !ok {
+				return
+			}
+			f.Sync()
+			f.Close()
+		}
+	}
+}

write_test.go

@@ -2,6 +2,7 @@ package zapper
 
 import (
 	"errors"
+	"fmt"
 	"path/filepath"
 	"testing"
 	"time"
@@ -22,11 +23,11 @@ func TestNewWriter(t *testing.T) {
 
 	syncLock := make(chan struct{}, 0)
 	go func() {
-		for i := 0; i <= 65; i++ {
-
+		for i := 0; i <= 240; i++ {
+			err:=errors.New("text string001")
 			Error("test cron zaper ",
 				zap.Int("int", 10),
-				zap.Error(errors.New("text string")),
+				zap.Error(fmt.Errorf("text string: %w",err)),
 				zap.String("msg", "val string1"),
 				zap.String("msg", "val string2"),
 				zap.String("msg", "val string3"),
@@ -51,4 +52,8 @@ func TestAbs(t *testing.T)  {
 		f,e:=filepath.Abs(s)
 		println(f,e)
 	}
+
+	f:="/Users/yymt/Documents/zaper/zapper2021-11-08 17:00:00.log"
+	f2,e:=filepath.Abs(f)
+	t.Log(f2,e)
 }