Golang并发安全日志系统设计与实现

channel

io.Writer

sync.Mutex

logEntry

channel

Info

Error

type LogEntry struct {
    Level   string
    Time    time.Time
    Message string
    Fields  map[string]interface{}
}

// logger.go
var logQueue chan LogEntry // 比如，容量设置为10000，可根据实际负载调整
var done chan struct{} // 用于通知写入器协程退出

func init() {
    logQueue = make(chan LogEntry, 10000)
    done = make(chan struct{})
    go startWriterLoop() // 启动日志写入协程
}

startWriterLoop

logQueue

logEntry

func startWriterLoop() {
    // 假设我们有一个io.Writer，比如一个文件句柄
    // logFile, _ := os.OpenFile("app.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
    // defer logFile.Close()
    // writer := bufio.NewWriter(logFile) // 使用bufio.Writer进行内部缓冲，减少系统调用

    // 假设这里是一个简单的标准输出写入
    writer := os.Stdout
    // 实际项目中会更复杂，比如文件轮转、网络发送等

    ticker := time.NewTicker(time.Second * 5) // 定期刷新缓冲区
    defer ticker.Stop()

    for {
        select {
        case entry := <-logQueue:
            // 在这里进行日志的格式化和实际写入
            formattedLog := formatLogEntry(entry) // 假设有这么一个格式化函数
            _, err := writer.Write([]byte(formattedLog + "\n"))
            if err != nil {
                // 写入错误处理，比如打印到stderr，或者尝试降级
                fmt.Fprintf(os.Stderr, "Error writing log: %v, log: %s\n", err, formattedLog)
            }

            // 考虑使用bufio.Writer时，需要定期Flush
            // if bw, ok := writer.(*bufio.Writer); ok {
            //     if bw.Buffered() > 4096 { // 积累一定量数据就刷新
            //         bw.Flush()
            //     }
            // }

        case <-ticker.C:
            // 定时刷新缓冲区，确保日志不会长时间停留在内存中
            // if bw, ok := writer.(*bufio.Writer); ok {
            //     bw.Flush()
            // }

        case <-done:
            // 收到关闭信号，清空剩余队列并退出
            // if bw, ok := writer.(*bufio.Writer); ok {
            //     bw.Flush()
            // }
            for { // 确保所有剩余日志被处理
                select {
                case entry := <-logQueue:
                    formattedLog := formatLogEntry(entry)
                    writer.Write([]byte(formattedLog + "\n"))
                default:
                    return // 队列已空，退出
                }
            }
        }
    }
}

// 示例的日志格式化函数
func formatLogEntry(entry LogEntry) string {
    // 实际项目中会使用更复杂的模板或JSON编码
    return fmt.Sprintf("[%s] %s %s - %v", entry.Level, entry.Time.Format(time.RFC3339), entry.Message, entry.Fields)
}

logQueue

select

func SendLog(level, msg string, fields map[string]interface{}) {
    entry := LogEntry{
        Level:   level,
        Time:    time.Now(),
        Message: msg,
        Fields:  fields,
    }

    select {
    case logQueue <- entry:
        // 成功发送
    default:
        // Channel已满，日志被丢弃。这是一种降级策略，避免阻塞主业务逻辑。
        // 可以在这里记录一个内部错误，表明日志系统过载。
        fmt.Fprintf(os.Stderr, "Log queue full, dropping log: %s\n", msg)
    }
}

// 提供便捷的日志级别方法
func Info(msg string, fields map[string]interface{}) { SendLog("INFO", msg, fields) }
func Error(msg string, fields map[string]interface{}) { SendLog("ERROR", msg, fields) }
// ... 其他级别

done

startWriterLoop

startWriterLoop

logQueue

// ShutdownLogSystem 在应用退出前调用，确保所有日志被写入
func ShutdownLogSystem() {
    close(done) // 通知写入器协程停止接收新日志并开始清理
    // 可以等待一段时间，确保写入器有足够时间处理剩余日志
    // time.Sleep(time.Second * 2) // 根据日志量和写入速度调整
}

channel

channel

channel

sync.Mutex

sync.Mutex

sync.RWMutex

Mutex

sync/atomic

sync/atomic

int32

int64

uint32

uint64

Pointer

atomic

sync.WaitGroup

WaitGroup

WaitGroup

channel

var writerWg sync.WaitGroup // 在init中初始化

func startWriterLoop() {
    writerWg.Add(1) // 启动时增加计数
    defer writerWg.Done() // 退出时减少计数

    // ... (之前的写入循环逻辑) ...
}

func ShutdownLogSystem() {
    close(done)
    writerWg.Wait() // 等待所有写入器协程完成
    // 此时可以确保所有日志都已处理完毕
}

channel

Mutex

atomic

WaitGroup