Golang并发处理文件IO优化示例

package main
<p>import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"strings"
"sync"
)</p><p>type FileContent struct {
Filename string
Content  []byte
Err      error
}</p><p>func readFile(filename string, ch chan<- FileContent) {
data, err := ioutil.ReadFile(filename)
ch <- FileContent{Filename: filename, Content: err, Err: nil}
}</p><p>func readFilesConcurrently(dir string) []FileContent {
var wg sync.WaitGroup
ch := make(chan FileContent, 10) // 缓冲channel避免阻塞
var results []FileContent</p><pre class='brush:php;toolbar:false;'>err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
    if err != nil {
        return err
    }
    if !info.IsDir() && strings.HasSuffix(info.Name(), ".txt") {
        wg.Add(1)
        go func(file string) {
            defer wg.Done()
            readFile(file, ch)
        }(path)
    }
    return nil
})

if err != nil {
    fmt.Printf("遍历目录出错: %v\n", err)
}

go func() {
    wg.Wait()
    close(ch)
}()

for result := range ch {
    if result.Err != nil {
        fmt.Printf("读取文件 %s 失败: %v\n", result.Filename, result.Err)
    } else {
        results = append(results, result)
    }
}

return results

func processLargeFile(filename string, chunkSize int64) error {
    file, err := os.Open(filename)
    if err != nil {
        return err
    }
    defer file.Close()
<pre class='brush:php;toolbar:false;'>fi, _ := file.Stat()
fileSize := fi.Size()
var wg sync.WaitGroup
ch := make(chan error, 10)

for i := int64(0); i*chunkSize < fileSize; i++ {
    start := i * chunkSize
    end := start + chunkSize
    if end > fileSize {
        end = fileSize
    }

    if start >= fileSize {
        break
    }

    wg.Add(1)
    go func(offset, length int64) {
        defer wg.Done()

        buffer := make([]byte, length)
        n, err := file.ReadAt(buffer, offset)
        if err != nil && err.Error() != "EOF" {
            ch <- fmt.Errorf("读取块失败 at %d: %v", offset, err)
            return
        }

        // 模拟处理：打印前10字节
        fmt.Printf("处理块 %d-%d: %q\n", offset, offset+int64(n), buffer[:min(n, 10)])

    }(start, end-start)
}

go func() {
    wg.Wait()
    close(ch)
}()

var errs []error
for err := range ch {
    errs = append(errs, err)
}

if len(errs) > 0 {
    return fmt.Errorf("处理过程中发生 %d 个错误", len(errs))
}
return nil

func readFilesWithWorkerPool(dir string, numWorkers int) []FileContent {
    jobs := make(chan string, 100)
    results := make(chan FileContent, 100)
    var wg sync.WaitGroup
    var files []string
<pre class='brush:php;toolbar:false;'>filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
    if !info.IsDir() && strings.HasSuffix(info.Name(), ".txt") {
        files = append(files, path)
    }
    return nil
})

for w := 0; w < numWorkers; w++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for filename := range jobs {
            data, err := ioutil.ReadFile(filename)
            results <- FileContent{Filename: filename, Content: data, Err: err}
        }
    }()
}

go func() {
    for _, f := range files {
        jobs <- f
    }
    close(jobs)
}()

go func() {
    wg.Wait()
    close(results)
}()

var finalResults []FileContent
for r := range results {
    if r.Err != nil {
        fmt.Printf("错误: %v\n", r.Err)
    } else {
        finalResults = append(finalResults, r)
    }
}

return finalResults