Go语言大文件处理：解密并发读取与性能优化策略

package main

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

// readLines 逐行读取文件内容
func readLines(filePath string) error {
    file, err := os.Open(filePath)
    if err != nil {
        return fmt.Errorf("无法打开文件: %w", err)
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        line := scanner.Text()
        // 这里可以进行简单的处理，例如打印或计数
        // fmt.Println(line)
        _ = line // 占位符，避免IDE警告
    }

    if err := scanner.Err(); err != nil {
        return fmt.Errorf("读取文件时发生错误: %w", err)
    }
    return nil
}

// createDummyFile 辅助函数：创建模拟文件
func createDummyFile(filename string, numLines int) {
    file, err := os.Create(filename)
    if err != nil {
        panic(err)
    }
    defer file.Close()
    writer := bufio.NewWriter(file)
    for i := 0; i < numLines; i++ {
        fmt.Fprintf(writer, "这是第 %d 行数据，用于测试文件读取。\n", i+1)
    }
    writer.Flush()
}

func main() {
    testFile := "large_file_sequential.txt"
    createDummyFile(testFile, 100000) // 创建一个包含10万行的模拟文件

    fmt.Printf("开始顺序读取文件 '%s'...\n", testFile)
    startTime := time.Now()
    if err := readLines(testFile); err != nil {
        fmt.Println(err)
    }
    fmt.Printf("文件顺序读取完成，耗时: %v\n", time.Since(startTime))

    // 清理模拟文件
    os.Remove(testFile)
}

package main

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

const (
    numWorkers = 4    // 并发处理的goroutine数量
    bufferSize = 1000 // channel缓冲区大小
)

// simulateHeavyProcessing 模拟耗时的数据处理函数
func simulateHeavyProcessing(line string) {
    // 模拟一些CPU密集型或I/O密集型操作
    time.Sleep(10 * time.Millisecond) // 模拟每行处理10毫秒
    // fmt.Printf("处理完成: %s\n", line) // 打印会增加I/O，此处注释掉
}

// processFileConcurrently 结合goroutines并发处理文件
func processFileConcurrently(filePath string) error {
    file, err := os.Open(filePath)
    if err != nil {
        return fmt.Errorf("无法打开文件: %w", err)
    }
    defer file.Close()

    lines := make(chan string, bufferSize) // 带缓冲的channel，用于传递行数据
    var wg sync.WaitGroup                  // 用于等待所有工作goroutine完成

    // 启动工作goroutine
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for line := range lines { // 从channel接收数据，直到channel关闭
                simulateHeavyProcessing(line)
                // fmt.Printf("Worker %d 处理了: %s\n", workerID, line)
            }
        }(i)
    }

    // 主goroutine负责读取文件并将行发送到channel
    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        lines <- scanner.Text() // 将读取到的行发送到channel
    }

    if err := scanner.Err(); err != nil {
        return fmt.Errorf("读取文件时发生错误: %w", err)
    }

    close(lines) // 关闭channel，通知工作goroutine没有更多数据了
    wg.Wait()    // 等待所有工作goroutine完成

    return nil
}

func main() {
    testFile := "large_file_concurrent.txt"
    // 使用与上一个示例相同的 createDummyFile 辅助函数
    createDummyFile(testFile, 5000) // 创建一个包含5千行的模拟文件，每行处理10ms，理论总处理时间50s

    fmt.Printf("开始并发处理文件 '%s'...\n", testFile)
    startTime := time.Now()
    if err := processFileConcurrently(testFile); err != nil {
        fmt.Println(err)
    }
    fmt.Printf("文件并发处理完成，耗时: %v\n", time.Since(startTime))

    // 清理模拟文件
    os.Remove(testFile)
}

// createDummyFile 辅助函数，与上一个示例相同，为避免重复此处省略，实际代码中需包含
/*
func createDummyFile(filename string, numLines int) {
    file, err := os.Create(filename)
    if err != nil {
        panic(err)
    }
    defer file.Close()
    writer := bufio.NewWriter(file)
    for i := 0; i < numLines; i++ {
        fmt.Fprintf(writer, "这是第 %d 行数据，用于测试文件读取。\n", i+1)
    }
    writer.Flush()
}
*/