Go语言并发编程中的OS线程限制与最佳实践

type AnalyzedPath struct {
    Path    string
    Content string // 假设分析结果包含内容
    Error   error
}

// GetPathsWorker 是一个工作函数，用于处理单个root并发送路径
func GetPathsWorker(root string, paths chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    // 模拟glob操作，实际中可能涉及文件系统遍历
    // 假设glob(root)返回一个字符串切片
    for _, path := range glob(root) { // glob(root) 是一个假设的函数
        paths <- path
    }
}

// GetPaths 使用工作池模式生成所有路径
func GetPaths(roots []string, numWorkers int) <-chan string {
    paths := make(chan string)
    var wg sync.WaitGroup

    go func() {
        // 创建一个buffered channel作为信号量，限制同时运行的goroutine数量
        // 这里可以直接使用WaitGroup来等待所有root处理完毕
        // 如果glob操作本身很快，也可以直接在单个goroutine中处理所有root
        // 但为了演示工作池，我们假设glob操作可能耗时

        // 限制同时处理root的goroutine数量
        sem := make(chan struct{}, numWorkers) 

        for _, root := range roots {
            wg.Add(1)
            sem <- struct{}{} // 获取一个信号量槽位
            go func(r string) {
                defer func() {
                    <-sem // 释放信号量槽位
                    wg.Done()
                }()
                // 模拟glob操作
                for _, p := range glob(r) {
                    paths <- p
                }
            }(root)
        }

        wg.Wait() // 等待所有root处理完成
        close(paths) // 所有路径都已发送，关闭通道
    }()
    return paths
}

// 假设的glob函数
func glob(root string) []string {
    // 实际的glob操作可能涉及os.ReadDir, filepath.Glob等
    // 这里简单模拟
    time.Sleep(50 * time.Millisecond) // 模拟耗时操作
    return []string{root + "/file1.txt", root + "/file2.txt"}
}

import (
    "fmt"
    "sync"
    "time" // 仅用于模拟耗时操作
)

// Analyze 是一个假设的分析函数
func Analyze(path string) AnalyzedPath {
    time.Sleep(100 * time.Millisecond) // 模拟耗时分析
    if path == "" { // 示例错误处理
        return AnalyzedPath{Path: path, Error: fmt.Errorf("empty path")}
    }
    return AnalyzedPath{Path: path, Content: "analyzed content for " + path}
}

// AnalyzeWorker 是一个工作函数，从输入通道读取路径，分析后发送到输出通道
func AnalyzeWorker(id int, paths <-chan string, analyzed chan<- AnalyzedPath, wg *sync.WaitGroup) {
    defer wg.Done()
    for path := range paths {
        result := Analyze(path)
        analyzed <- result
    }
}

// AnalyzePaths 使用工作池模式分析路径
func AnalyzePaths(paths <-chan string, numWorkers int) <-chan AnalyzedPath {
    analyzed := make(chan AnalyzedPath)
    var wg sync.WaitGroup

    // 启动固定数量的worker goroutine
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go AnalyzeWorker(i, paths, analyzed, &wg)
    }

    // 启动一个goroutine来等待所有worker完成，然后关闭输出通道
    go func() {
        wg.Wait() // 等待所有worker goroutine完成
        close(analyzed) // 所有结果都已发送，关闭输出通道
    }()

    return analyzed
}

func main() {
    patterns := []string{"/data/root1", "/data/root2", "/data/root3"}

    // 设置GetPaths的并发度，例如，同时处理2个root
    pathsChan := GetPaths(patterns, 2) 

    // 设置AnalyzePaths的并发度，例如，同时有4个goroutine进行分析
    analyzedChan := AnalyzePaths(pathsChan, 4) 

    for result := range analyzedChan {
        if result.Error != nil {
            fmt.Printf("Error analyzing %s: %v\n", result.Path, result.Error)
        } else {
            fmt.Printf("Analyzed: %s, Content: %s\n", result.Path, result.Content)
        }
    }
    fmt.Println("All analysis complete.")
}