Go语言中高效处理大型文件：理解I/O瓶颈与并发策略

package main

import (
    "bufio"
    "fmt"
    "os"
    "time"
)

func readAndProcessFileBuffered(filePath string) {
    file, err := os.Open(filePath)
    if err != nil {
        fmt.Printf("Error opening file: %v\n", err)
        return
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    lineCount := 0
    startTime := time.Now()

    for scanner.Scan() {
        line := scanner.Text()
        // 这里模拟对每一行数据的处理
        _ = line 
        lineCount++
    }

    if err := scanner.Err(); err != nil {
        fmt.Printf("Error reading file: %v\n", err)
    }

    fmt.Printf("Processed %d lines in %s (Buffered Reading)\n", lineCount, time.Since(startTime))
}

func main() {
    // 创建一个大型测试文件 (如果不存在)
    testFilePath := "large_test_file.txt"
    if _, err := os.Stat(testFilePath); os.IsNotExist(err) {
        fmt.Println("Creating a large test file...")
        createLargeTestFile(testFilePath, 1000000) // 100万行
        fmt.Println("Test file created.")
    }

    readAndProcessFileBuffered(testFilePath)
}

// 辅助函数：创建一个大型测试文件
func createLargeTestFile(filePath string, numLines int) {
    file, err := os.Create(filePath)
    if err != nil {
        panic(err)
    }
    defer file.Close()

    writer := bufio.NewWriter(file)
    for i := 0; i < numLines; i++ {
        fmt.Fprintf(writer, "This is line number %d of a very large file.\n", i+1)
    }
    writer.Flush()
}

package main

import (
    "bufio"
    "fmt"
    "os"
    "runtime"
    "sync"
    "time"
)

// 模拟每行数据的处理逻辑
func processLine(line string) {
    // 模拟CPU密集型操作，例如复杂的计算、解析、编码等
    // 实际应用中，这里会是业务逻辑
    time.Sleep(time.Microsecond * 10) // 模拟耗时操作
    _ = line // 避免未使用变量警告
}

func readAndProcessFileConcurrent(filePath string, numWorkers int) {
    file, err := os.Open(filePath)
    if err != nil {
        fmt.Printf("Error opening file: %v\n", err)
        return
    }
    defer file.Close()

    lineChannel := make(chan string, 1000) // 带缓冲的通道，防止生产者阻塞
    var wg sync.WaitGroup
    lineCount := 0
    startTime := time.Now()

    // 生产者 goroutine：读取文件并将行发送到通道
    wg.Add(1)
    go func() {
        defer wg.Done()
        scanner := bufio.NewScanner(file)
        for scanner.Scan() {
            lineChannel <- scanner.Text()
        }
        if err := scanner.Err(); err != nil {
            fmt.Printf("Error reading file in producer: %v\n", err)
        }
        close(lineChannel) // 读取完毕，关闭通道
    }()

    // 消费者 goroutines：从通道接收行并处理
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for line := range lineChannel {
                processLine(line)
                // 注意：lineCount的增量操作需要同步，但在这个例子中，我们只在主goroutine中统计总数
                // 如果需要在消费者中统计，需要使用原子操作或互斥锁
            }
        }()
    }

    // 等待所有goroutines完成
    wg.Wait()

    // 重新打开文件以获取总行数，或者在生产者中统计
    // 这里为了简化示例，我们假设文件读取后可以知道总行数
    // 实际应用中，生产者在发送时可以计数，或者在消费者处理完后汇总
    fileStats, _ := os.Stat(filePath)
    if fileStats != nil {
        // 简单的模拟，实际应通过计数器获取准确的已处理行数
        // 这里为了演示，假设所有行都被处理了
        tempFile, _ := os.Open(filePath)
        tempScanner := bufio.NewScanner(tempFile)
        for tempScanner.Scan() {
            lineCount++
        }
        tempFile.Close()
    }


    fmt.Printf("Processed %d lines in %s with %d workers (Concurrent Processing)\n", lineCount, time.Since(startTime), numWorkers)
}

func main() {
    testFilePath := "large_test_file.txt"
    // 确保测试文件存在
    if _, err := os.Stat(testFilePath); os.IsNotExist(err) {
        fmt.Println("Creating a large test file...")
        createLargeTestFile(testFilePath, 1000000) // 100万行
        fmt.Println("Test file created.")
    }

    // 使用CPU核心数作为默认工作协程数
    numWorkers := runtime.NumCPU()
    fmt.Printf("Using %d CPU cores for workers.\n", numWorkers)
    readAndProcessFileConcurrent(testFilePath, numWorkers)
}

// 辅助函数：创建一个大型测试文件 (同上)
func createLargeTestFile(filePath string, numLines int) {
    file, err := os.Create(filePath)
    if err != nil {
        panic(err)
    }
    defer file.Close()

    writer := bufio.NewWriter(file)
    for i := 0; i < numLines; i++ {
        fmt.Fprintf(writer, "This is line number %d of a very large file.\n", i+1)
    }
    writer.Flush()
}