Golang如何优化日志输出 异步写入与缓冲队列方案

bufio.Writer

log.Println

fmt.Fprintf

logChan chan []byte

fmt.Fprintln(file, msg)

logChan <- []byte(msg)

logChan

bytes.Buffer

bytes.Buffer

// 简化示例，实际应用需要更完善的错误处理和优雅关闭
package main

import (
    "bytes"
    "fmt"
    "log"
    "os"
    "sync"
    "time"
)

// LogEntry 是日志消息的结构
type LogEntry struct {
    Level string
    Msg   string
    Time  time.Time
}

// LogWriter 负责异步写入
type LogWriter struct {
    logChan   chan LogEntry
    file      *os.File
    buffer    *bytes.Buffer // 内部缓冲
    flushSize int           // 达到此大小则刷新
    flushFreq time.Duration // 达到此频率则刷新
    stopChan  chan struct{}
    wg        sync.WaitGroup
}

func NewLogWriter(filePath string, bufferSize int, flushFreq time.Duration) (*LogWriter, error) {
    f, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
    if err != nil {
        return nil, fmt.Errorf("failed to open log file: %w", err)
    }

    lw := &LogWriter{
        logChan:   make(chan LogEntry, 10000), // 缓冲通道，防止瞬时峰值阻塞
        file:      f,
        buffer:    bytes.NewBuffer(make([]byte, 0, bufferSize)),
        flushSize: bufferSize,
        flushFreq: flushFreq,
        stopChan:  make(chan struct{}),
    }
    lw.wg.Add(1)
    go lw.run() // 启动日志写入goroutine
    return lw, nil
}

func (lw *LogWriter) run() {
    defer lw.wg.Done()
    ticker := time.NewTicker(lw.flushFreq)
    defer ticker.Stop()

    for {
        select {
        case entry := <-lw.logChan:
            // 格式化日志并写入内部缓冲区
            fmt.Fprintf(lw.buffer, "[%s] %s %s\n", entry.Level, entry.Time.Format("2006-01-02 15:04:05"), entry.Msg)
            if lw.buffer.Len() >= lw.flushSize {
                lw.flush()
            }
        case <-ticker.C:
            // 定时刷新，防止日志长时间积压在内存
            if lw.buffer.Len() > 0 {
                lw.flush()
            }
        case <-lw.stopChan:
            // 收到停止信号，处理完剩余日志后退出
            for len(lw.logChan) > 0 { // 确保通道中所有日志被处理
                entry := <-lw.logChan
                fmt.Fprintf(lw.buffer, "[%s] %s %s\n", entry.Level, entry.Time.Format("2006-01-02 15:04:05"), entry.Msg)
            }
            lw.flush() // 最终刷新
            lw.file.Close()
            log.Println("LogWriter stopped and file closed.")
            return
        }
    }
}

func (lw *LogWriter) flush() {
    if lw.buffer.Len() == 0 {
        return
    }
    _, err := lw.file.Write(lw.buffer.Bytes())
    if err != nil {
        log.Printf("Error writing logs to file: %v\n", err)
    }
    lw.buffer.Reset() // 清空缓冲区
}

// Log 供外部调用的日志记录方法
func (lw *LogWriter) Log(level, msg string) {
    select {
    case lw.logChan <- LogEntry{Level: level, Msg: msg, Time: time.Now()}:
        // Log entry sent successfully
    default:
        // 通道已满，可以考虑丢弃日志或阻塞
        // 这里选择丢弃，避免阻塞业务逻辑
        log.Printf("Log channel full, dropping log: [%s] %s\n", level, msg)
    }
}

// Stop 停止日志写入器，并确保所有日志被写入
func (lw *LogWriter) Stop() {
    close(lw.stopChan)
    lw.wg.Wait() // 等待run goroutine退出
}

func main() {
    writer, err := NewLogWriter("app.log", 4096, 5*time.Second) // 4KB缓冲区，5秒刷新一次
    if err != nil {
        log.Fatalf("Failed to create log writer: %v", err)
    }
    defer writer.Stop() // 确保程序退出时日志被刷入

    for i := 0; i < 100000; i++ {
        writer.Log("INFO", fmt.Sprintf("Processing request %d", i))
        if i%10000 == 0 {
            time.Sleep(10 * time.Millisecond) // 模拟业务处理
        }
    }
    log.Println("All logs sent to channel. Waiting for flush...")
    time.Sleep(2 * time.Second) // 确保有时间刷新最后一部分日志
}

LogWriter

logChan

run

Log

logChan

run

logChan

channel

LogWriter

bytes.Buffer