Golang并发日志写入安全实现方法

sync.Mutex

channel

sync.Mutex

package main

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

// 定义一个全局的互斥锁和文件句柄
var (
    logFile *os.File
    mu      sync.Mutex
)

func init() {
    var err error
    logFile, err = os.OpenFile("app.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
    if err != nil {
        log.Fatalf("无法打开日志文件: %v", err)
    }
    // 确保程序退出时关闭文件
    // defer logFile.Close() // 注意：这里不能用defer，因为init函数会先执行完
}

// WriteLogFunc 模拟一个日志写入函数
func WriteLogFunc(id int, message string) {
    mu.Lock() // 获取锁
    defer mu.Unlock() // 确保函数退出时释放锁

    // 实际写入操作
    _, err := logFile.WriteString(fmt.Sprintf("[%s] Goroutine %d: %s\n", time.Now().Format("15:04:05.000"), id, message))
    if err != nil {
        fmt.Printf("Goroutine %d 写入日志失败: %v\n", id, err)
    }
}

func main() {
    // 确保在main函数退出时关闭文件
    defer func() {
        if logFile != nil {
            logFile.Close()
            fmt.Println("日志文件已关闭。")
        }
    }()

    var wg sync.WaitGroup
    for i := 0; i < 5; i++ { // 启动5个Goroutine并发写入
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := 0; j < 10; j++ {
                WriteLogFunc(id, fmt.Sprintf("这是第 %d 条日志", j+1))
                time.Sleep(time.Millisecond * time.Duration(id*10+10)) // 模拟一些工作负载和随机延迟
            }
        }(i)
    }
    wg.Wait()
    fmt.Println("所有Goroutine写入完成。")
}

WriteLogFunc

mu.Lock()

defer mu.Unlock()

sync.Mutex

channel

channel

package main

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

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

// logChan 用于接收所有日志消息的通道
var logChan chan LogEntry
// logFileChannelWriter 是实际的日志文件句柄
var logFileChannelWriter *os.File
// doneChan 用于通知日志写入Goroutine停止
var doneChan chan struct{}
// wgWriter 用于等待日志写入Goroutine结束
var wgWriter sync.WaitGroup

func init() {
    var err error
    logFileChannelWriter, err = os.OpenFile("app_channel.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
    if err != nil {
        log.Fatalf("无法打开日志文件: %v", err)
    }

    // 初始化通道，可以设置缓冲区大小
    logChan = make(chan LogEntry, 1000) // 缓冲区大小1000
    doneChan = make(chan struct{})

    // 启动一个Goroutine专门负责日志写入
    wgWriter.Add(1)
    go logWriterGoroutine()
}

// logWriterGoroutine 负责从通道读取日志并写入文件
func logWriterGoroutine() {
    defer wgWriter.Done()
    defer func() {
        if logFileChannelWriter != nil {
            logFileChannelWriter.Close()
            fmt.Println("Channel日志文件已关闭。")
        }
    }()

    for {
        select {
        case entry := <-logChan:
            // 实际写入操作
            _, err := logFileChannelWriter.WriteString(
                fmt.Sprintf("[%s] [%s] Goroutine %d: %s\n",
                    entry.Timestamp.Format("15:04:05.000"), entry.Level, entry.GoroutineID, entry.Message))
            if err != nil {
                fmt.Printf("日志写入失败: %v\n", err)
            }
        case <-doneChan:
            // 收到停止信号，处理完通道中剩余的日志
            fmt.Println("收到停止信号，正在处理剩余日志...")
            for {
                select {
                case entry := <-logChan:
                    _, err := logFileChannelWriter.WriteString(
                        fmt.Sprintf("[%s] [%s] Goroutine %d: %s\n",
                            entry.Timestamp.Format("15:04:05.000"), entry.Level, entry.GoroutineID, entry.Message))
                    if err != nil {
                        fmt.Printf("剩余日志写入失败: %v\n", err)
                    }
                default:
                    fmt.Println("所有日志已处理完毕。")
                    return // 通道已空，退出
                }
            }
        }
    }
}

// SendLog 供其他Goroutine调用的日志发送函数
func SendLog(id int, level, message string) {
    select {
    case logChan <- LogEntry{Timestamp: time.Now(), Level: level, Message: message, GoroutineID: id}:
        // 成功发送
    default:
        // 通道已满，可以考虑丢弃日志或阻塞等待
        // 在这里，我们选择丢弃，避免阻塞生产者
        fmt.Printf("Goroutine %d: 日志通道已满，丢弃日志: %s\n", id, message)
    }
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := 0; j < 10; j++ {
                SendLog(id, "INFO", fmt.Sprintf("这是第 %d 条日志", j+1))
                time.Sleep(time.Millisecond * time.Duration(id*10+5)) // 模拟一些工作负载
            }
        }(i)
    }

    wg.Wait() // 等待所有生产者Goroutine完成
    fmt.Println("所有生产者Goroutine写入完成。")

    // 通知日志写入Goroutine停止
    close(doneChan)
    wgWriter.Wait() // 等待日志写入Goroutine完成
    fmt.Println("程序退出。")
}

channel