Go语言中处理海量数据：避免嵌套Goroutine，构建高效通道流水线

// 伪代码示例
func processFolder(folderPath string) {
    files := readFiles(folderPath)
    for _, file := range files {
        go do1(file) // 为每个文件启动一个Goroutine
    }
}

func do1(file File) {
    lines := readLines(file)
    for _, line := range lines {
        go do2(line) // 为文件中的每行再启动一个Goroutine
    }
}

func do2(line Line) {
    // 执行具体的行处理逻辑
    // do_something
}

// 伪代码示例
func processFolderAndLines(folderPath string) {
    files := readFiles(folderPath)
    for _, file := range files {
        lines := readLines(file)
        for _, line := range lines {
            go do_something(line) // 为每个文件的每行都启动一个Goroutine
        }
    }
}

func do_something(line Line) {
    // 执行具体的行处理逻辑
}

package main

import (
    "fmt"
    "io/ioutil"
    "log"
    "path/filepath"
    "strings"
    "sync"
    "time"
)

// 定义通道容量，用于流量控制
const (
    fileChanCapacity = 100
    lineChanCapacity = 1000
    numLineProducers = 4  // 示例：启动4个Goroutine解析文件并生成行
    numLineWorkers   = 8  // 示例：启动8个Goroutine处理行数据
)

// 模拟文件结构和行结构
type File struct {
    Path string
    Name string
}

type Line struct {
    FilePath string
    Content  string
    LineNum  int
}

func main() {
    // 创建通道
    fileChan := make(chan File, fileChanCapacity)
    lineChan := make(chan Line, lineChanCapacity)
    // resultChan := make(chan ResultType, ...) // 如果需要收集处理结果

    var wg sync.WaitGroup

    // --- 阶段1: 文件生产者 ---
    // 启动一个Goroutine负责遍历文件夹并发送文件信息
    wg.Add(1)
    go func() {
        defer wg.Done()
        defer close(fileChan) // 文件发送完毕后关闭文件通道
        folderPath := "./test_data" // 假设存在一个test_data文件夹
        files, err := ioutil.ReadDir(folderPath)
        if err != nil {
            log.Printf("Error reading directory %s: %v", folderPath, err)
            return
        }
        for _, f := range files {
            if !f.IsDir() {
                filePath := filepath.Join(folderPath, f.Name())
                fileChan <- File{Path: filePath, Name: f.Name()}
                fmt.Printf("Produced file: %s\n", f.Name())
            }
        }
    }()

    // --- 阶段2: 行生产者 ---
    // 启动多个Goroutine从fileChan接收文件，解析行，并发送到lineChan
    for i := 0; i < numLineProducers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for file := range fileChan { // 循环直到fileChan被关闭且清空
                fmt.Printf("[LineProducer %d] Processing file: %s\n", workerID, file.Name)
                content, err := ioutil.ReadFile(file.Path)
                if err != nil {
                    log.Printf("Error reading file %s: %v", file.Path, err)
                    continue
                }
                lines := strings.Split(string(content), "\n")
                for lineNum, lineContent := range lines {
                    if strings.TrimSpace(lineContent) != "" {
                        lineChan <- Line{FilePath: file.Path, Content: lineContent, LineNum: lineNum + 1}
                    }
                }
            }
            fmt.Printf("[LineProducer %d] Finished.\n", workerID)
        }(i)
    }

    // 启动一个Goroutine等待所有文件生产者完成，然后关闭lineChan
    wg.Add(1)
    go func() {
        defer wg.Done()
        // 等待所有文件生产者Goroutine完成 (即fileChan被关闭且所有数据被消费)
        // 注意：这里的wg.Wait()会等待所有wg.Add(1)对应的wg.Done()
        // 实际上，更精确的做法是有一个单独的WaitGroup用于监控LineProducers
        // 但为了简化示例，我们知道当fileChan关闭且被消费完后，LineProducers会退出
        // 所以这里可以简单地等待一段时间，或者使用更复杂的信号机制
        // 这里我们直接依赖于fileChan的关闭，并在所有LineProducers退出后关闭lineChan
        // 更好的做法是：创建一个新的WaitGroup for LineProducers, 并在所有LineProducers退出后关闭lineChan
        // 考虑到这个示例，我们让主goroutine在所有生产者完成后关闭lineChan
        // 更好的模式是：
        // 1. 文件生产者关闭 fileChan
        // 2. 一个单独的 goroutine 监听 LineProducers 的 wg，当它们都退出时关闭 lineChan
        // 为了简化，我们使用一个稍微不那么优雅但能工作的方案：
        // 等待足够的时间，确保所有文件都被处理，或者使用一个更精细的协调机制
        // 这里我们依赖于 main goroutine 的 wg.Wait() 来确保所有生产者都退出了。
        // 实际上，我们应该在所有 numLineProducers 都退出后才关闭 lineChan。
        // 最直接的方法是创建一个新的 WaitGroup 专门给 LineProducers。
        // 这里为了演示，我们假设在文件生产者关闭后，lineChan 最终会被关闭。
        // 实际操作中，应该有一个独立的 Goroutine 来监控 LineProducers 的完成状态。
        // 临时方案：等待所有 LineProducers 退出
        // 重新设计这部分，确保lineChan在所有LineProducers完成后关闭。
        // 创建一个新的 WaitGroup 来等待 LineProducers
        var lineProducerWg sync.WaitGroup
        for i := 0; i < numLineProducers; i++ {
            lineProducerWg.Add(1)
            go func(workerID int) {
                defer lineProducerWg.Done()
                for file := range fileChan {
                    // ... (同上，处理文件并发送行到 lineChan)
                    fmt.Printf("[LineProducer %d] Processing file: %s\n", workerID, file.Name)
                    content, err := ioutil.ReadFile(file.Path)
                    if err != nil {
                        log.Printf("Error reading file %s: %v", file.Path, err)
                        continue
                    }
                    lines := strings.Split(string(content), "\n")
                    for lineNum, lineContent := range lines {
                        if strings.TrimSpace(lineContent) != "" {
                            lineChan <- Line{FilePath: file.Path, Content: lineContent, LineNum: lineNum + 1}
                        }
                    }
                }
                fmt.Printf("[LineProducer %d] Finished.\n", workerID)
            }(i)
        }

        // 确保文件生产者已经启动并关闭了 fileChan
        // 这里需要一个机制来等待文件生产者完成，但不能让它阻塞主goroutine
        // 简单起见，我们让主goroutine等待所有wg.Done()
        // 正确的模式是：
        // 1. fileProducer 启动并关闭 fileChan
        // 2. lineProducers 从 fileChan 读取，处理完后，如果 fileChan 关闭且为空，它们会退出
        // 3. 一个单独的 goroutine 监听 lineProducers 的退出，当所有 lineProducers 退出后，关闭 lineChan
        // 为了简化，我们让主 goroutine 等待所有 wg.Done()，并在所有生产者完成后关闭 lineChan。
        // 这是一个经典的扇出-扇入模式，需要谨慎处理通道的关闭时机。

        // 正确的关闭 lineChan 策略：
        // 1. 等待所有 fileChan 的消费者 (即 lineProducers) 完成。
        // 2. 一旦所有 lineProducers 完成，关闭 lineChan。
        // 我们可以通过一个额外的 Goroutine 来监听 lineProducerWg 的完成。
        go func() {
            lineProducerWg.Wait() // 等待所有 LineProducers 完成
            close(lineChan)       // 所有 LineProducers 都退出了，可以安全关闭 lineChan
            fmt.Println("Line channel closed.")
        }()

        // 注意：上面的 numLineProducers 循环应该放在这里，而不是 main 的直接子 Goroutine
        // 否则 lineProducerWg 无法正确计数。
        // 重新组织：
        // 文件生产者在 main Goroutine 中启动，并关闭 fileChan。
        // 多个行生产者 Goroutine 在 main Goroutine 中启动，并添加到 lineProducerWg。
        // 一个单独的 Goroutine 等待 lineProducerWg 完成并关闭 lineChan。
        // 多个行处理器 Goroutine 在 main Goroutine 中启动，并添加到主 wg。

        // --- 阶段3: 行处理器 ---
        // 启动多个Goroutine从lineChan接收行数据并处理
        for i := 0; i < numLineWorkers; i++ {
            wg.Add(1)
            go func(workerID int) {
                defer wg.Done()
                for line := range lineChan { // 循环直到lineChan被关闭且清空
                    // 模拟行处理逻辑
                    time.Sleep(5 * time.Millisecond) // 模拟耗时操作
                    fmt.Printf("[LineWorker %d] Processed line %d from %s: %s\n", workerID, line.LineNum, filepath.Base(line.FilePath), line.Content)
                    // 如果有结果，可以发送到 resultChan
                }
                fmt.Printf("[LineWorker %d] Finished.\n", workerID)
            }(i)
        }

        // 等待所有 Goroutine 完成
        wg.Wait()
        fmt.Println("All processing finished.")
    }() // 这里的匿名函数是为了将 lineProducerWg 的逻辑放在正确的位置

    // 重新组织 main 函数以清晰地展示流程
    // 创建通道
    fileChan = make(chan File, fileChanCapacity)
    lineChan = make(chan Line, lineChanCapacity)

    var mainWg sync.WaitGroup // 主 WaitGroup 来等待所有 Goroutine

    // --- 阶段1: 文件生产者 ---
    mainWg.Add(1)
    go func() {
        defer mainWg.Done()
        defer close(fileChan) // 文件发送完毕后关闭文件通道
        folderPath := "./test_data"
        // 确保 test_data 文件夹存在并有文件，用于测试
        // 例如：
        // mkdir test_data
        // echo "line1\nline2" > test_data/file1.txt
        // echo "lineA\nlineB" > test_data/file2.txt
        files, err := ioutil.ReadDir(folderPath)
        if err != nil {
            log.Printf("Error reading directory %s: %v. Please create it and add some files for testing.", folderPath, err)
            return
        }
        for _, f := range files {
            if !f.IsDir() {
                filePath := filepath.Join(folderPath, f.Name())
                fileChan <- File{Path: filePath, Name: f.Name()}
                // fmt.Printf("Produced file: %s\n", f.Name()) // 避免过多输出
            }
        }
    }()

    // --- 阶段2: 行生产者 ---
    var lineProducerWg sync.WaitGroup // 专门用于等待行生产者
    for i := 0; i < numLineProducers; i++ {
        lineProducerWg.Add(1)
        mainWg.Add(1) // 也添加到主 WaitGroup
        go func(workerID int) {
            defer lineProducerWg.Done()
            defer mainWg.Done()
            for file := range fileChan {
                // fmt.Printf("[LineProducer %d] Processing file: %s\n", workerID, file.Name) // 避免过多输出
                content, err := ioutil.ReadFile(file.Path)
                if err != nil {
                    log.Printf("Error reading file %s: %v", file.Path, err)
                    continue
                }
                lines := strings.Split(string(content), "\n")
                for lineNum, lineContent := range lines {
                    if strings.TrimSpace(lineContent) != "" {
                        lineChan <- Line{FilePath: file.Path, Content: lineContent, LineNum: lineNum + 1}
                    }
                }
            }
            // fmt.Printf("[LineProducer %d] Finished.\n", workerID) // 避免过多输出
        }(i)
    }

    // 监听 lineProducerWg，当所有行生产者完成时关闭 lineChan
    mainWg.Add(1)
    go func() {
        defer mainWg.Done()
        lineProducerWg.Wait() // 等待所有 LineProducers 完成
        close(lineChan)       // 所有 LineProducers 都退出了，可以安全关闭 lineChan
        fmt.Println("Line channel closed.")
    }()

    // --- 阶段3: 行处理器 ---
    for i := 0; i < numLineWorkers; i++ {
        mainWg.Add(1)
        go func(workerID int) {
            defer mainWg.Done()
            for line := range lineChan {
                // 模拟行处理逻辑
                time.Sleep(5 * time.Millisecond) // 模拟耗时操作
                // fmt.Printf("[LineWorker %d] Processed line %d from %s: %s\n", workerID, line.LineNum, filepath.Base(line.FilePath), line.Content) // 避免过多输出
            }
            // fmt.Printf("[LineWorker %d] Finished.\n", workerID) // 避免过多输出
        }(i)
    }

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