1
// Stage 1: 生成器：产生数据源
2
func generate(nums ...int) <-chan int {
3
    out := make(chan int)
4
    go func() {
5
        defer close(out)
6
        for _, n := range nums {
7
            out <- n
8
        }
9
    }()
10
    return out
11
}
12

13
// Stage 2: 平方运算：转换数据
14
func square(in <-chan int) <-chan int {
15
    out := make(chan int)
16
    go func() {
17
        defer close(out)
18
        for n := range in {
19
            out <- n * n
20
        }
21
    }()
22
    return out
23
}
24

25
// Stage 3: 过滤：筛选符合条件的值
26
func filter(in <-chan int, predicate func(int) bool) <-chan int {
27
    out := make(chan int)
28
    go func() {
29
        defer close(out)
30
        for n := range in {
31
            if predicate(n) {
32
                out <- n
33
            }
34
        }
35
    }()
36
    return out
37
}
38

39
// 使用示例
40
func main() {
41
    // 构建流水线
42
    nums := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
43
    squares := square(nums)
44
    evens := filter(squares, func(n int) bool { return n > 25 })
45

46
    // 消费结果
47
    for n := range evens {
48
        fmt.Println(n)
49
    }
50
}

1
func generateContext(ctx context.Context, nums ...int) <-chan int {
2
    out := make(chan int)
3
    go func() {
4
        defer close(out)
5
        for _, n := range nums {
6
            select {
7
            case out <- n:
8
            case <-ctx.Done():
9
                return // 取消时退出
10
            }
11
        }
12
    }()
13
    return out
14
}
15

16
func squareContext(ctx context.Context, in <-chan int) <-chan int {
17
    out := make(chan int)
18
    go func() {
19
        defer close(out)
20
        for {
21
            select {
22
            case n, ok := <-in:
23
                if !ok {
24
                    return
25
                }
26
                select {
27
                case out <- n * n:
28
                case <-ctx.Done():
29
                    return
30
                }
31
            case <-ctx.Done():
32
                return
33
            }
34
        }
35
    }()
36
    return out
37
}

1
// 主动丢弃策略：channel 满时跳过当前数据
2
func dropOnFull(in <-chan int) <-chan int {
3
    out := make(chan int, 10) // 有限缓冲
4
    go func() {
5
        defer close(out)
6
        for n := range in {
7
            select {
8
            case out <- n:
9
            default:
10
                // channel 满时丢弃，避免上游阻塞
11
                log.Printf("dropped: %d", n)
12
            }
13
        }
14
    }()
15
    return out
16
}

1
// 并行执行多个 worker
2
func fanOut(worker func(<-chan Task) <-chan Result, jobs <-chan Task, workers int) []<-chan Result {
3
    results := make([]<-chan Result, workers)
4
    for i := 0; i < workers; i++ {
5
        results[i] = worker(jobs)
6
    }
7
    return results
8
}
9

10
func worker(jobs <-chan Task) <-chan Result {
11
    out := make(chan Result)
12
    go func() {
13
        defer close(out)
14
        for job := range jobs {
15
            out <- process(job)
16
        }
17
    }()
18
    return out
19
}

1
// 合并多个 channel 的结果
2
func fanIn(ctx context.Context, channels ...<-chan Result) <-chan Result {
3
    out := make(chan Result)
4

5
    var wg sync.WaitGroup
6
    wg.Add(len(channels))
7

8
    // 为每个输入 channel 启动一个 goroutine
9
    for _, ch := range channels {
10
        go func(c <-chan Result) {
11
            defer wg.Done()
12
            for result := range c {
13
                select {
14
                case out <- result:
15
                case <-ctx.Done():
16
                    return
17
                }
18
            }
19
        }(ch)
20
    }
21

22
    // 等待所有 channel 关闭后关闭输出
23
    go func() {
24
        wg.Wait()
25
        close(out)
26
    }()
27

28
    return out
29
}

1
func fetchURLs(urls []string) ([]Page, error) {
2
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
3
    defer cancel()
4

5
    // 创建任务 channel
6
    jobs := make(chan string, len(urls))
7
    for _, url := range urls {
8
        jobs <- url
9
    }
10
    close(jobs)
11

12
    // Fan-out：启动多个 worker 并行抓取
13
    numWorkers := min(5, len(urls))
14
    results := fanOut(fetchWorker, jobs, numWorkers)
15

16
    // Fan-in：合并结果
17
    merged := fanIn(ctx, results...)
18

19
    // 收集结果
20
    var pages []Page
21
    for result := range merged {
22
        if result.Err != nil {
23
            return nil, result.Err
24
        }
25
        pages = append(pages, result.Page)
26
    }
27

28
    return pages, nil
29
}
30

31
func fetchWorker(jobs <-chan string) <-chan FetchResult {
32
    out := make(chan FetchResult)
33
    go func() {
34
        defer close(out)
35
        client := http.Client{Timeout: 10 * time.Second}
36
        for url := range jobs {
37
            resp, err := client.Get(url)
38
            if err != nil {
39
                out <- FetchResult{Err: err}
40
                continue
41
            }
42
            body, _ := io.ReadAll(resp.Body)
43
            resp.Body.Close()
44
            out <- FetchResult{Page: Page{URL: url, Content: string(body)}}
45
        }
46
    }()
47
    return out
48
}

1
type WorkerPool struct {
2
    tasks   chan Task
3
    results chan Result
4
    workers int
5
    wg      sync.WaitGroup
6
}
7

8
func NewWorkerPool(workers, taskCapacity int) *WorkerPool {
9
    return &WorkerPool{
10
        tasks:   make(chan Task, taskCapacity),
11
        results: make(chan Result, taskCapacity),
12
        workers: workers,
13
    }
14
}
15

16
func (p *WorkerPool) Start() {
17
    for i := 0; i < p.workers; i++ {
18
        p.wg.Add(1)
19
        go p.worker(i)
20
    }
21
}
22

23
func (p *WorkerPool) worker(id int) {
24
    defer p.wg.Done()
25
    for task := range p.tasks {
26
        result := process(task)
27
        p.results <- result
28
    }
29
}
30

31
func (p *WorkerPool) Submit(task Task) {
32
    p.tasks <- task
33
}
34

35
func (p *WorkerPool) Stop() {
36
    close(p.tasks)
37
    p.wg.Wait()
38
    close(p.results)
39
}
40

41
func (p *WorkerPool) Results() <-chan Result {
42
    return p.results
43
}

1
type CancellableWorkerPool struct {
2
    tasks   chan Task
3
    results chan Result
4
    workers int
5
    ctx     context.Context
6
    cancel  context.CancelFunc
7
    wg      sync.WaitGroup
8
}
9

10
func NewCancellableWorkerPool(ctx context.Context, workers, capacity int) *CancellableWorkerPool {
11
    childCtx, cancel := context.WithCancel(ctx)
12
    return &CancellableWorkerPool{
13
        tasks:   make(chan Task, capacity),
14
        results: make(chan Result, capacity),
15
        workers: workers,
16
        ctx:     childCtx,
17
        cancel:  cancel,
18
    }
19
}
20

21
func (p *CancellableWorkerPool) Start() {
22
    for i := 0; i < p.workers; i++ {
23
        p.wg.Add(1)
24
        go p.worker(i)
25
    }
26
}
27

28
func (p *CancellableWorkerPool) worker(id int) {
29
    defer p.wg.Done()
30
    for {
31
        select {
32
        case task, ok := <-p.tasks:
33
            if !ok {
34
                return
35
            }
36
            result := p.processWithCancel(task)
37
            select {
38
            case p.results <- result:
39
            case <-p.ctx.Done():
40
                return
41
            }
42
        case <-p.ctx.Done():
43
            return
44
        }
45
    }
46
}
47

48
func (p *CancellableWorkerPool) processWithCancel(task Task) Result {
49
    // 在处理过程中也检查取消信号
50
    done := make(chan Result, 1)
51
    go func() {
52
        done <- process(task)
53
    }()
54

55
    select {
56
    case result := <-done:
57
        return result
58
    case <-p.ctx.Done():
59
        return Result{Err: p.ctx.Err()}
60
    }
61
}
62

63
func (p *CancellableWorkerPool) Cancel() {
64
    p.cancel()
65
}
66

67
func (p *CancellableWorkerPool) Stop() {
68
    close(p.tasks)
69
    p.wg.Wait()
70
    close(p.results)
71
}

1
type Context interface {
2
    // 返回 context 被取消的截止时间
3
    Deadline() (deadline time.Time, ok bool)
4

5
    // 返回一个 channel，当 context 被取消时关闭
6
    Done() <-chan struct{}
7

8
    // 返回取消原因
9
    Err() error
10

11
    // 获取与 context 关联的值
12
    Value(key any) any
13
}

1
func operation(ctx context.Context) error {
2
    // 启动子任务
3
    ctx, cancel := context.WithCancel(ctx)
4
    defer cancel() // 确保资源释放
5

6
    // 启动多个子任务
7
    results := make(chan Result, 3)
8
    for i := 0; i < 3; i++ {
9
        go func(id int) {
10
            results <- subOperation(ctx, id)
11
        }(i)
12
    }
13

14
    // 等待第一个结果或取消
15
    select {
16
    case result := <-results:
17
        cancel() // 取消其他子任务
18
        return result.Err
19
    case <-ctx.Done():
20
        return ctx.Err()
21
    }
22
}
23

24
func subOperation(ctx context.Context, id int) Result {
25
    // 定期检查取消信号
26
    for i := 0; i < 100; i++ {
27
        select {
28
        case <-ctx.Done():
29
            return Result{Err: ctx.Err()}
30
        default:
31
            // 执行工作
32
            time.Sleep(10 * time.Millisecond)
33
        }
34
    }
35
    return Result{Value: id}
36
}

1
// 正确：不要将 Context 存储在结构体中（少数例外：HTTP Handler）
2
type Handler struct {
3
    // ctx 不应该作为字段
4
}

1
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
2
    // 创建带超时的 context
3
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
4
    defer cancel()
5

6
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
7
    if err != nil {
8
        return nil, err
9
    }
10

11
    resp, err := http.DefaultClient.Do(req)
12
    if err != nil {
13
        if errors.Is(err, context.DeadlineExceeded) {
14
            return nil, fmt.Errorf("request timed out: %w", err)
15
        }
16
        return nil, err
17
    }
18

19
    return resp, nil
20
}

1
func simpleTimeout(timeout time.Duration) error {
2
    result := make(chan error, 1)
3

4
    go func() {
5
        result <- doWork()
6
    }()
7

8
    select {
9
    case err := <-result:
10
        return err
11
    case <-time.After(timeout):
12
        return fmt.Errorf("operation timed out after %v", timeout)
13
    }
14
}

1
func multiLayerTimeout() error {
2
    // 外层：总超时 30 秒
3
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
4
    defer cancel()
5

6
    // 阶段 1：数据库查询，超时 10 秒
7
    dbCtx, dbCancel := context.WithTimeout(ctx, 10*time.Second)
8
    data, err := queryDB(dbCtx)
9
    dbCancel()
10
    if err != nil {
11
        return fmt.Errorf("database query: %w", err)
12
    }
13

14
    // 阶段 2：外部 API 调用，超时 15 秒
15
    apiCtx, apiCancel := context.WithTimeout(ctx, 15*time.Second)
16
    result, err := callAPI(apiCtx, data)
17
    apiCancel()
18
    if err != nil {
19
        return fmt.Errorf("api call: %w", err)
20
    }
21

22
    return nil
23
}

1
import "golang.org/x/sync/errgroup"
2

3
func fetchAll(urls []string) ([]Page, error) {
4
    g, ctx := errgroup.WithContext(context.Background())
5
    pages := make([]Page, len(urls))
6

7
    for i, url := range urls {
8
        i, url := i, url // 捕获循环变量
9
        g.Go(func() error {
10
            // 如果其他 goroutine 出错，ctx 会被取消
11
            page, err := fetchPage(ctx, url)
12
            if err != nil {
13
                return err
14
            }
15
            pages[i] = page
16
            return nil
17
        })
18
    }
19

20
    // 等待所有任务完成，返回第一个错误（如果有）
21
    if err := g.Wait(); err != nil {
22
        return nil, err
23
    }
24

25
    return pages, nil
26
}

1
func fetchWithLimit(urls []string, maxConcurrent int) ([]Page, error) {
2
    g, ctx := errgroup.WithContext(context.Background())
3

4
    // 使用信号量限制并发数
5
    sem := make(chan struct{}, maxConcurrent)
6
    pages := make([]Page, len(urls))
7

8
    for i, url := range urls {
9
        i, url := i, url
10
        sem <- struct{}{} // 获取信号量
11

12
        g.Go(func() error {
13
            defer func() { <-sem }() // 释放信号量
14

15
            page, err := fetchPage(ctx, url)
16
            if err != nil {
17
                return err
18
            }
19
            pages[i] = page
20
            return nil
21
        })
22
    }
23

24
    if err := g.Wait(); err != nil {
25
        return nil, err
26
    }
27

28
    return pages, nil
29
}

1
func monitorGoroutines() {
2
    ticker := time.NewTicker(5 * time.Second)
3
    for range ticker.C {
4
        count := runtime.NumGoroutine()
5
        log.Printf("current goroutines: %d", count)
6
        if count > 100 {
7
            log.Printf("WARNING: too many goroutines!")
8
        }
9
    }
10
}

1
import _ "net/http/pprof"
2

3
func main() {
4
    go func() {
5
        http.ListenAndServe("localhost:6060", nil)
6
    }()
7
    // 访问 http://localhost:6060/debug/pprof/goroutine?debug=1
8
    // 查看 goroutine 堆栈
9
}