package loggerx import ( "io" "log" "time" ) // 异步队列容量,跟随实例 const asyncQueueSize = 1000 // Close 等待异步日志落盘的最长时间 const closeDrainTimeout = 5 * time.Second // 写入,需要判断同步还是异步 func (l *Logger) write(event string, b []byte) (n int, err error) { if l.toAsync(event, b) { return len(b), nil } return l.store(event, b) } // 实际的存储 // 必须满足 io.Writer 契约:成功时 n == len(b),否则调用方(log / io.MultiWriter) // 会认为发生了短写并把日志吞掉 func (l *Logger) store(event string, b []byte) (n int, err error) { return l.storeTo(l.channel, event, b) } // storeTo 按指定 channel 落盘(异步消费协程用,它的 l.channel 是空的) func (l *Logger) storeTo(channel, event string, b []byte) (n int, err error) { if l.option.isPrintFile { // 串行化写入:句柄缓冲区不是并发安全的, // 异步消费协程与同步调用可能同时写同一个句柄 l.writeMu.Lock() n, err = l.storeFileTo(channel, event, b) l.writeMu.Unlock() if err != nil { return 0, err } } // 驱动(控制台等)的写入失败不改变对调用方的契约 _, _ = l.writeDrivers(b) return len(b), nil } // storeFile 写入日志文件 func (l *Logger) storeFile(event string, b []byte) (int, error) { return l.storeFileTo(l.channel, event, b) } // storeFileTo 把日志写进指定 channel 对应的文件 func (l *Logger) storeFileTo(channel, event string, b []byte) (int, error) { f, err := l.getFileTo(channel, event) if err != nil { return 0, err } // 按大小切割:这条日志会把当前文件写超上限时,先归档当前文件再换新文件 if limit := l.option.sizeSplit; limit > 0 && f.full(limit, len(b)) { nf, rerr := l.rollFileTo(channel, event, f) if rerr == nil { f = nf } // 归档失败就继续往原文件写:宁可文件超一点,也不能把日志丢了 } n, err := f.Write(b) if err == nil && n < len(b) { err = io.ErrShortWrite } if err == nil { return n, nil } // 写入失败:丢弃这个句柄,落到磁盘后按最新文件名重开一次再写 // 只重试一次,避免原实现在短写时反复重开文件 l.discardFileTo(channel, event, f) if nf, nerr := l.getFileTo(channel, event); nerr == nil { if n2, err2 := nf.Write(b); err2 == nil && n2 == len(b) { return n2, nil } } return 0, err } // rollFileTo 归档当前文件并返回新文件句柄 func (l *Logger) rollFileTo(channel, event string, f *logFile) (*logFile, error) { key := fileKey{channel: channel, event: event} // 先把缓冲清空再关句柄,保证归档内容完整 if err := f.Close(); err != nil { return nil, err } // 归档基名直接用句柄记下的 baseName:绝不去猜文件名尾部的 _N 是不是归档序号 // (按小时切割出来的名字本身就长这样:2026/09/13/06_info.log) path, err := l.archive(f.baseName, f.fileName) if err != nil { return nil, err } l.mu.Lock() if cur, ok := l.filePath[key]; ok && cur == f { delete(l.filePath, key) } l.mu.Unlock() // 新文件带走递增序号,避免覆盖刚归档出去的同名文件 nf, err := l.openNumberedFile(key, l.nextIndex()) if err != nil { return nil, err } l.mu.Lock() l.filePath[key] = nf l.mu.Unlock() // 后台压缩,不阻塞写入 l.scheduleCompress(path) return nf, nil } // 写入额外的驱动(控制台 / 自定义 writer) func (l *Logger) writeDrivers(b []byte) (int, error) { if len(l.option.drivers) == 0 { return 0, nil } return io.MultiWriter(l.option.drivers...).Write(b) } // discardFile 关闭文件并从缓存中移除,使下次写入重新打开 func (l *Logger) discardFile(event string, f *logFile) { l.discardFileTo(l.channel, event, f) } // discardFileTo 关闭并移除指定 channel 的文件句柄 func (l *Logger) discardFileTo(channel, event string, f *logFile) { _ = f.Close() l.mu.Lock() defer l.mu.Unlock() key := fileKey{channel: channel, event: event} if cur, ok := l.filePath[key]; ok && cur == f { delete(l.filePath, key) } } // 异步队列的任务 // Channel 必须跟着任务走:消费协程持有的是「创建队列时的那个 logger」, // 它的 channel 是空的。如果消费时用 logger 自己的 channel, // 所有 channel 的日志都会落进根目录文件(曾经就是这么错的) type cacheData struct { Channel string Event string Data []byte } // toAsync 尝试异步写入 // 返回 true 表示已经交给异步队列,返回 false 表示需要同步写入 func (l *Logger) toAsync(event string, b []byte) bool { if l.writeType == writeTypeSync || // 指定同步模式 (l.writeType == writeTypeDefault && l.option.writeType != writeTypeAsync) { // 默认同步模式 return false } // 整段「检查开关 + 入队」都在同一把锁内:Close 也拿这把锁来停止投递, // 这样 Close 拿到锁时就能确定「要么这条还没入队、要么已经完整入队」。 // 否则消费协程可能先看到空队列就退出,把还在路上的这条日志整条丢掉。 // // wg 用来告诉 drainAsync「投递临界区里还有人」:Close 必须等这些写入 // 真正入队之后才能关闭队列,否则向已关闭的 channel 发送会 panic。 // 队列满时这里会阻塞,但消费者是独立 goroutine 且不需要这把锁,不会死锁 l.async.mu.Lock() if l.async.closed { l.async.mu.Unlock() // 已开始关闭:退化为同步写入 return false } if l.async.ch == nil { l.async.ch = make(chan cacheData, asyncQueueSize) go l.asyncWorker(l.async.ch) } l.async.begin() ch := l.async.ch l.async.mu.Unlock() defer l.async.end() // 必须在解锁之后 Done,保证 begin/end 覆盖整段投递 ch <- cacheData{Channel: l.channel, Event: event, Data: b} return true } // asyncWorker 消费异步队列,直到队列被关闭 func (l *Logger) asyncWorker(q chan cacheData) { defer close(l.workerDone) defer func() { if r := recover(); r != nil { log.Println("loggerx: 异步写入协程异常:", r) } }() for val := range q { // 按任务里记的 channel 落盘,避免多个 channel 的日志混到根目录 _, _ = l.storeTo(val.Channel, val.Event, val.Data) } } // drainAsync 停止投递、关闭队列,并等消费协程把剩余任务全部写完 func (l *Logger) drainAsync() { // 持锁关闭投递:此后 toAsync 一律退化为同步写入 l.async.mu.Lock() l.async.closed = true q := l.async.ch l.async.mu.Unlock() if q == nil { // 从未启用过异步写入,没有后台 goroutine 需要等 return } // 关键:必须等「已经进入投递临界区」的写入全部入队之后才能关队列。 // 这里能看到 wg 已经是 0,就说明没有写入还卡在投递路径上, // 否则 close(q) 之后它们再发就会 panic: send on closed channel l.async.wg.Wait() // 关闭队列并等消费者把缓冲里的任务全部处理完。 // 这一步必须真的等到 workerDone:否则 Close 会在消费协程还在写缓冲时 // 就刷盘并关闭文件,最后几条日志会连着句柄一起丢掉 close(q) select { case <-l.workerDone: case <-time.After(closeDrainTimeout): log.Println("loggerx: 等待异步日志落盘超时") } }