Go语言并发模型与OS线程限制：高效管理Goroutine的策略

package main

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

// 假设的AnalyzedPath结构和Analyze函数
type AnalyzedPath struct {
    Path  string
    Size  int
    Error error
}

func Analyze(path string) AnalyzedPath {
    // 模拟耗时操作，可能涉及I/O阻塞
    time.Sleep(100 * time.Millisecond)
    if path == "error_path" {
        return AnalyzedPath{Path: path, Error: fmt.Errorf("simulated error")}
    }
    return AnalyzedPath{Path: path, Size: len(path) * 10}
}

// 改进后的AnalyzePaths，使用工作池限制并发
func AnalyzePathsBounded(paths <-chan string, workerCount int) <-chan AnalyzedPath {
    analyzed := make(chan AnalyzedPath)
    var wg sync.WaitGroup

    // 启动固定数量的工作Goroutine
    for i := 0; i < workerCount; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for path := range paths { // 从输入通道接收任务
                analyzed <- Analyze(path) // 将结果发送到输出通道
            }
        }()
    }

    // 启动一个Goroutine来等待所有工作Goroutine完成，然后关闭输出通道
    go func() {
        wg.Wait()
        close(analyzed)
    }()

    return analyzed
}

// 模拟GetPaths函数
func GetPaths(roots []string) <-chan string {
    globbed := make(chan string)
    go func() {
        defer close(globbed) // 确保通道最终关闭
        for _, root := range roots {
            // 模拟文件系统遍历
            for i := 0; i < 5; i++ { // 每个root生成5个路径
                globbed <- fmt.Sprintf("%s/file_%d.txt", root, i)
            }
        }
    }()
    return globbed
}

func main() {
    patterns := []string{"/data/dir1", "/data/dir2", "/data/dir3"}
    inputPaths := GetPaths(patterns)

    // 设置并发度为5
    const maxConcurrentAnalyzers = 5
    results := AnalyzePathsBounded(inputPaths, maxConcurrentAnalyzers)

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

package main

import (
    "fmt"
    "time"
)

// 定义一些类型
type SomeType1 int
type SomeType2 string

// 生产者Goroutine
func generator(ch1 chan<- SomeType1, ch2 chan<- SomeType2) {
    defer func() {
        close(ch1) // 生产者完成所有数据生产后关闭数据通道
        ch2 <- "generator_finished" // 发送生产者完成信号
    }()

    for i := 0; i < 10; i++ {
        fmt.Printf("Generator producing: %d\n", i)
        ch1 <- SomeType1(i)
        time.Sleep(50 * time.Millisecond) // 模拟生产耗时
    }
}

// 消费者Goroutine
func processor(ch1 <-chan SomeType1, ch2 chan<- SomeType2) {
    defer func() {
        ch2 <- "processor_finished" // 发送消费者完成信号
    }()

    for value := range ch1 { // 从数据通道读取数据，直到通道关闭
        fmt.Printf("Processor consuming: %d\n", value)
        time.Sleep(100 * time.Millisecond) // 模拟处理耗时
    }
    fmt.Println("Processor finished consuming all items.")
}

func main() {
    ch1 := make(chan SomeType1) // 数据通道
    ch2 := make(chan SomeType2, 2) // 信号通道，缓冲2，用于接收两个Goroutine的完成信号

    go generator(ch1, ch2)
    go processor(ch1, ch2)

    // main Goroutine等待所有工作Goroutine完成
    // 通过从信号通道接收两个完成信号来判断
    <-ch2
    <-ch2
    fmt.Println("All Goroutines finished, main exiting.")
}