Go语言大文件解析：利用Channel实现多级并发任务的优雅调度与资源控制

// 伪代码示例
func processFolder(folderPath string) {
    for _, file := range readFiles(folderPath) {
        go func(f File) {
            processFile(f)
        }(file)
    }
}

func processFile(file File) {
    for _, line := range readLines(file) {
        go func(l Line) {
            doSomething(l) // 对每行数据进行处理
        }(line)
    }
}

func doSomething(line Line) {
    // 实际的行处理逻辑
}

// 伪代码示例
func processFolderFlat(folderPath string) {
    for _, file := range readFiles(folderPath) {
        for _, line := range readLines(file) {
            go func(l Line) {
                doSomething(l) // 对每行数据进行处理
            }(line)
        }
    }
}

// fileChan 用于传递文件
fileChan := make(chan File, bufferSize) // 适当的缓冲区大小

// 启动一个goroutine发送文件
go func() {
    defer close(fileChan) // 发送完毕后关闭Channel
    for _, file := range folder { // 假设 folder 是一个文件列表
        fileChan <- file
    }
}()

// lineChan 用于传递行数据
lineChan := make(chan Line, bufferSize) // 适当的缓冲区大小

// 启动一个或多个goroutine从fileChan接收文件，并提取行
// 这里以一个goroutine为例，实际可启动多个并行处理文件
go func() {
    defer close(lineChan) // 所有文件处理完毕后关闭Channel
    for file := range fileChan { // 持续从fileChan接收文件
        for _, line := range file.ReadLines() { // 假设file有ReadLines方法
            lineChan <- line
        }
    }
}()

// 启动多个goroutine并行处理行数据
numWorkers := runtime.NumCPU() // 根据CPU核心数或实际需求设定工作协程数量
var wg sync.WaitGroup // 用于等待所有工作协程完成

for i := 0; i < numWorkers; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for line := range lineChan { // 持续从lineChan接收行数据
            // 实际的行处理逻辑
            processLine(line)
        }
    }()
}

// 在主goroutine中等待所有行处理器完成
wg.Wait()

package main

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

// 模拟文件和行的数据结构
type File struct {
    Name    string
    Content []string
}

type Line string

// 模拟读取文件内容
func (f File) ReadLines() []Line {
    lines := make([]Line, len(f.Content))
    for i, c := range f.Content {
        lines[i] = Line(c)
    }
    return lines
}

// 模拟行处理函数
func processLine(line Line) {
    // 模拟耗时操作
    time.Sleep(10 * time.Millisecond)
    // fmt.Printf("Processed: %s\n", line) // 打印会影响性能，实际应用中谨慎
}

func main() {
    // 模拟文件夹中的大量文件
    folder := []File{
        {Name: "file1.txt", Content: []string{"line1-1", "line1-2", "line1-3", "line1-4"}},
        {Name: "file2.txt", Content: []string{"line2-1", "line2-2", "line2-3", "line2-4"}},
        {Name: "file3.txt", Content: []string{"line3-1", "line3-2", "line3-3", "line3-4"}},
        {Name: "file4.txt", Content: []string{"line4-1", "line4-2", "line4-3", "line4-4"}},
        // 更多文件...
    }

    // 定义Channel
    fileChan := make(chan File, 5) // 文件Channel，缓冲区5
    lineChan := make(chan Line, 10) // 行Channel，缓冲区10

    var wg sync.WaitGroup // 用于等待所有goroutine完成

    // 1. 文件分发器
    wg.Add(1)
    go func() {
        defer wg.Done()
        defer close(fileChan) // 发送完毕后关闭fileChan
        for _, file := range folder {
            fileChan <- file
            fmt.Printf("Dispatched file: %s\n", file.Name)
        }
    }()

    // 2. 行提取器 (可以启动多个，这里以1个为例)
    wg.Add(1)
    go func() {
        defer wg.Done()
        defer close(lineChan) // 所有文件处理完毕后关闭lineChan
        for file := range fileChan { // 从fileChan接收文件
            fmt.Printf("Extracting lines from: %s\n", file.Name)
            for _, line := range file.ReadLines() {
                lineChan <- line
            }
        }
    }()

    // 3. 行处理器 (启动多个工作协程)
    numWorkers := runtime.NumCPU() // 通常设置为CPU核心数
    fmt.Printf("Starting %d line processors...\n", numWorkers)
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for line := range lineChan { // 从lineChan接收行数据
                // fmt.Printf("Worker %d processing: %s\n", workerID, line)
                processLine(line)
            }
            fmt.Printf("Worker %d finished.\n", workerID)
        }(i)
    }

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