Go语言并发文件处理：避免嵌套Goroutine陷阱与高效资源管理策略

package main

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

// fileProducer 负责遍历指定文件夹，将文件路径发送到fileChan通道。
// 完成后关闭fileChan。
func fileProducer(folderPath string, fileChan chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("生产者：开始扫描文件夹 %s\n", folderPath)

    // 模拟文件遍历，实际应用中应使用os.ReadDir或filepath.Walk
    // 为了简化示例，我们创建一些虚拟文件
    virtualFiles := []string{"doc1.txt", "doc2.txt", "doc3.txt", "doc4.txt", "doc5.txt"}
    for _, fileName := range virtualFiles {
        filePath := folderPath + "/" + fileName
        // 模拟文件内容写入，以便后续读取
        content := fmt.Sprintf("这是文件 %s 的第一行。\n这是文件 %s 的第二行。\n这是文件 %s 的第三行。", fileName, fileName, fileName)
        err := ioutil.WriteFile(filePath, []byte(content), 0644)
        if err != nil {
            log.Printf("生产者：写入虚拟文件 %s 失败：%v\n", filePath, err)
            continue
        }
        fileChan <- filePath // 将文件路径发送到通道
        fmt.Printf("生产者：发送文件路径 %s\n", filePath)
    }
    close(fileChan) // 所有文件路径都已发送，关闭通道
    fmt.Println("生产者：所有文件路径已发送，fileChan已关闭。")
}

// lineExtractor 负责从fileChan接收文件路径，读取文件内容，并将每行数据发送到lineChan通道。
// 完成后关闭lineChan。
func lineExtractor(fileChan <-chan string, lineChan chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("行提取器：已启动。")
    for filePath := range fileChan { // 从fileChan接收文件路径
        fmt.Printf("行提取器：正在处理文件 %s\n", filePath)
        content, err := ioutil.ReadFile(filePath) // 读取文件内容
        if err != nil {
            log.Printf("行提取器：读取文件 %s 失败：%v\n", filePath, err)
            continue
        }
        lines := strings.Split(string(content), "\n") // 按行分割
        for _, line := range lines {
            if strings.TrimSpace(line) != "" { // 忽略空行
                lineChan <- line // 将每行数据发送到lineChan
                fmt.Printf("行提取器：发送行数据 '%s'\n", line)
            }
        }
        time.Sleep(time.Millisecond * 100) // 模拟文件读取和解析的耗时
    }
    close(lineChan) // 所有文件都已处理，关闭通道
    fmt.Println("行提取器：所有行已提取，lineChan已关闭。")
}

// lineProcessor 负责从lineChan接收行数据并执行实际的处理逻辑。
// 这是工作池中的一个工作Goroutine。
func lineProcessor(id int, lineChan <-chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("处理器 %d：已启动。\n", id)
    for line := range lineChan { // 从lineChan接收行数据
        fmt.Printf("处理器 %d：正在处理行 '%s'\n", id, line)
        time.Sleep(time.Millisecond * 200) // 模拟耗时操作
        // 在这里执行实际的业务逻辑，例如数据清洗、存储到数据库等
    }
    fmt.Printf("处理器 %d：任务完成，退出。\n", id)
}

func main() {
    // 创建一个用于等待所有Goroutine完成的WaitGroup
    var wg sync.WaitGroup

    // 创建通道：
    // fileChan 用于传递文件路径，缓冲大小为5，防止生产者过快。
    fileChan := make(chan string, 5)
    // lineChan 用于传递文件中的行数据，缓冲大小为100，应对行数据突发。
    lineChan := make(chan string, 100)

    // 1. 启动文件路径生产者
    wg.Add(1)
    go fileProducer("temp_folder", fileChan, &wg) // 假设文件在"temp_folder"下

    // 2. 启动文件内容分解器（行提取器）
    wg.Add(1)
    go lineExtractor(fileChan, lineChan, &wg)

    // 3. 启动多个行数据处理器（工作池）
    numWorkers := 3 // 控制并发度，可以根据CPU核心数和任务类型调整
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go lineProcessor(i, lineChan, &wg)
    }

    // 等待所有Goroutine完成任务
    wg.Wait()
    fmt.Println("主程序：所有任务已完成，程序退出。")
}