Golang如何实现高效的并发日志收集 结合Channel与异步写入实践

package main

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

// LogEntry 定义日志条目的结构
type LogEntry struct {
    Timestamp time.Time
    Level     string
    Message   string
}

// LogCollector 定义日志收集器
type LogCollector struct {
    logChan   chan LogEntry
    file      *os.File
    wg        sync.WaitGroup
    closeOnce sync.Once
}

// NewLogCollector 创建一个新的日志收集器
func NewLogCollector(filePath string, bufferSize int) (*LogCollector, error) {
    file, err := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
    if err != nil {
        return nil, err
    }

    collector := &LogCollector{
        logChan:   make(chan LogEntry, bufferSize),
        file:      file,
        wg:        sync.WaitGroup{},
        closeOnce: sync.Once{},
    }

    collector.wg.Add(1)
    go collector.writeLogsToFile()

    return collector, nil
}

// CollectLog 收集日志
func (lc *LogCollector) CollectLog(entry LogEntry) {
    lc.logChan <- entry
}

// writeLogsToFile 从通道读取日志并写入文件
func (lc *LogCollector) writeLogsToFile() {
    defer lc.wg.Done()
    defer lc.file.Close()

    for entry := range lc.logChan {
        logLine := fmt.Sprintf("%s [%s] %s\n", entry.Timestamp.Format(time.RFC3339), entry.Level, entry.Message)
        _, err := lc.file.WriteString(logLine)
        if err != nil {
            // 处理写入错误，例如，记录到错误日志
            fmt.Fprintf(os.Stderr, "Error writing log: %v\n", err)
        }
    }
}

// Close 关闭日志收集器
func (lc *LogCollector) Close() error {
    var err error
    lc.closeOnce.Do(func() {
        close(lc.logChan) // 关闭通道，通知写入协程退出
        lc.wg.Wait()       // 等待写入协程完成
        err = lc.file.Close() // 关闭文件
    })
    return err
}

func main() {
    collector, err := NewLogCollector("app.log", 1000)
    if err != nil {
        log.Fatalf("Failed to create log collector: %v", err)
    }
    defer collector.Close()

    // 模拟并发日志生成
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := 0; j < 100; j++ {
                entry := LogEntry{
                    Timestamp: time.Now(),
                    Level:     "INFO",
                    Message:   fmt.Sprintf("Goroutine %d: Log message %d", id, j),
                }
                collector.CollectLog(entry)
                time.Sleep(time.Millisecond * 10) // 模拟一些工作
            }
        }(i)
    }

    wg.Wait() // 等待所有goroutine完成
    fmt.Println("All logs collected.")
}