95 lines
1.9 KiB
Go
95 lines
1.9 KiB
Go
package logger
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// asyncBufferSize 异步写盘队列长度。队列满时丢弃日志，优先保护抢购链路。
|
|
const asyncBufferSize = 8192
|
|
|
|
// AsyncWriter 异步日志写入器，队列满时丢弃日志以保护主链路。
|
|
type AsyncWriter struct {
|
|
mu sync.RWMutex
|
|
dst io.Writer
|
|
ch chan []byte
|
|
wg sync.WaitGroup
|
|
closed uint32
|
|
dropped uint64
|
|
}
|
|
|
|
// NewAsyncWriter 创建异步日志写入器。
|
|
func NewAsyncWriter(dst io.Writer, buffer int) *AsyncWriter {
|
|
if buffer <= 0 {
|
|
buffer = 1024
|
|
}
|
|
w := &AsyncWriter{
|
|
dst: dst,
|
|
ch: make(chan []byte, buffer),
|
|
}
|
|
w.wg.Add(1)
|
|
go w.run()
|
|
return w
|
|
}
|
|
|
|
// Write 异步写日志；队列满时丢弃，避免抢购请求被磁盘 I/O 阻塞。
|
|
func (w *AsyncWriter) Write(p []byte) (int, error) {
|
|
if w == nil {
|
|
return len(p), nil
|
|
}
|
|
w.mu.RLock()
|
|
defer w.mu.RUnlock()
|
|
if atomic.LoadUint32(&w.closed) == 1 {
|
|
return len(p), nil
|
|
}
|
|
buf := append([]byte(nil), p...)
|
|
select {
|
|
case w.ch <- buf:
|
|
default:
|
|
atomic.AddUint64(&w.dropped, 1)
|
|
}
|
|
return len(p), nil
|
|
}
|
|
|
|
// Dropped 返回因队列满被丢弃的日志条数。
|
|
func (w *AsyncWriter) Dropped() uint64 {
|
|
if w == nil {
|
|
return 0
|
|
}
|
|
return atomic.LoadUint64(&w.dropped)
|
|
}
|
|
|
|
func (w *AsyncWriter) run() {
|
|
defer w.wg.Done()
|
|
for p := range w.ch {
|
|
_, _ = w.dst.Write(p)
|
|
}
|
|
if dropped := atomic.LoadUint64(&w.dropped); dropped > 0 {
|
|
notice := formatLogLine(Entry{
|
|
Time: time.Now().Format(entryTimeLayout),
|
|
Level: LevelWarn,
|
|
Category: "日志",
|
|
Message: fmt.Sprintf("日志队列已丢弃 %d 条，已优先保护抢购链路", dropped),
|
|
at: time.Now(),
|
|
})
|
|
_, _ = w.dst.Write([]byte(notice))
|
|
}
|
|
}
|
|
|
|
// Close 关闭写入器，等待队列清空。
|
|
func (w *AsyncWriter) Close() error {
|
|
if w == nil {
|
|
return nil
|
|
}
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
if atomic.CompareAndSwapUint32(&w.closed, 0, 1) {
|
|
close(w.ch)
|
|
w.wg.Wait()
|
|
}
|
|
return nil
|
|
}
|