并发模式（worker pool / 扇入扇出 / pipeline）

package main

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

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs { // range 自动感知 jobs 关闭
        time.Sleep(50 * time.Millisecond) // 模拟干活
        results <- j * 2
        fmt.Printf("工人 %d 处理了任务 %d\n", id, j)
    }
}

func main() {
    const numJobs, numWorkers = 10, 3
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    var wg sync.WaitGroup
    // 启动固定数量的工人
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // 主 goroutine 派发任务（提前派完）
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs) // 任务派发完，关掉任务队列 → workers 的 range 结束

    wg.Wait()   // 全部工人收工
    close(results) // 结果也关，方便 range 收集

    for r := range results {
        fmt.Println("结果:", r)
    }
}

package main

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

func main() {
    jobs := make(chan int, 10)

    // 扇出：两个消费组同时消费同一 jobs channel
    var wg sync.WaitGroup
    consume := func(name string) {
        defer wg.Done()
        for j := range jobs {
            fmt.Printf("%s 处理 %d\n", name, j)
            time.Sleep(30 * time.Millisecond)
        }
    }

    wg.Add(2)
    go consume("消费者A")
    go consume("消费者B")

    // 生产者喷出任务
    for i := 1; i <= 8; i++ {
        jobs <- i
    }
    close(jobs)

    wg.Wait()
    fmt.Println("扇出完成：8 个任务被 2 个消费者分掉了")
}

package main

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

func producer(name string, count int, out chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 1; i <= count; i++ {
        out <- fmt.Sprintf("%s-%d", name, i)
        time.Sleep(20 * time.Millisecond)
    }
}

func main() {
    results := make(chan string, 12)

    // 三个生产者都往同一个 results 里塞（扇入）
    var wg sync.WaitGroup
    wg.Add(3)
    go producer("苹果", 4, results, &wg)
    go producer("香蕉", 4, results, &wg)
    go producer("梨子", 4, results, &wg)

    // 关掉 producers 后统一关 results
    go func() {
        wg.Wait()
        close(results) // 全部生产者收工 → 消费方 range 才能结束
    }()

    // 需求方一条线收齐所有结果
    for r := range results {
        fmt.Println("收到:", r)
    }
    fmt.Println("扇入收敛完成")
}

package main

import (
    "fmt"
)

// stage 1: 生成数字
func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

// stage 2: 乘 2
func multiply(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * 2
        }
        close(out)
    }()
    return out
}

// stage 3: 加 1
func addOne(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n + 1
        }
        close(out)
    }()
    return out
}

func main() {
    // 流水线组装：gen → multiply → addOne
    pipeline := addOne(multiply(gen(1, 2, 3, 4)))

    for v := range pipeline {
        fmt.Println("流水线输出:", v) // (n*2+1): 3, 5, 7, 9
    }
}

package main

import (
    "fmt"
    "sync"
)

func main() {
    in := make(chan int, 10)
    out := make(chan string, 10)

    // 两个并行的「处理 stage」都写 out（扇入式的流水线宽度=2）
    var wg sync.WaitGroup
    for w := 1; w <= 2; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for n := range in {
                out <- fmt.Sprintf("stage%d: %d*10=%d", id, n, n*10)
            }
        }(w)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    for i := 1; i <= 5; i++ {
        in <- i
    }
    close(in)

    for r := range out {
        fmt.Println(r)
    }
}

package main

import (
    "fmt"
    "time"
)

func main() {
    // 令牌滴答：每 200ms 放行一次
    limiter := time.NewTicker(200 * time.Millisecond)
    defer limiter.Stop()

    requests := []int{1, 2, 3, 4, 5}
    for _, r := range requests {
        <-limiter.C // 令牌不到，这里就阻塞 → 天然限速
        fmt.Println("放行请求:", r, "at", time.Now().Format("15:04:05.000"))
    }

    // 带 burst 的限速：预填 3 个令牌，之后按 200ms 补一个
    bursty := make(chan struct{}, 3)
    for i := 0; i < 3; i++ {
        bursty <- struct{}{} // 预填（burst）
    }
    go func() {
        tick := time.NewTicker(200 * time.Millisecond)
        defer tick.Stop()
        for t := range tick.C {
            _ = t
            bursty <- struct{}{} // 低频补令牌
        }
    }()

    for i := 1; i <= 5; i++ {
        <-bursty // 前 3 个立刻放行（有 burst），后 2 个要等补令牌
        fmt.Println("突发请求", i, "放行")
    }
}

package main

import (
    "errors"
    "fmt"
    "time"
)

type Result struct {
    Value int
    Err   error
}

func worker(in <-chan int, out chan<- Result) {
    for n := range in {
        if n == 3 { // 模拟「这个任务会失败」
            out <- Result{Err: errors.New("task 3 数据非法")}
            continue
        }
        out <- Result{Value: n * 2}
    }
}

func main() {
    tasks := make(chan int, 5)
    results := make(chan Result, 5)

    go worker(tasks, results)

    for i := 1; i <= 5; i++ {
        tasks <- i
    }
    close(tasks)

    for i := 0; i < 5; i++ {
        r := <-results
        if r.Err != nil {
            fmt.Println("发现错误:", r.Err)
            continue
        }
        fmt.Println("结果:", r.Value)
    }
    _ = time.Second
}

package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    var g errgroup.Group

    for i := 1; i <= 4; i++ {
        i := i
        g.Go(func() error { // g.Go 就驱动一个 goroutine
            select {
            case <-time.After(time.Duration(i) * 100 * time.Millisecond):
                if i == 2 {
                    return errors.New("stage 2 失败")
                }
                return nil
            case <-ctx.Done():
                return ctx.Err() // 别的任务失败 → 取消信号 → 快速退出
            }
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Println("errgroup 捕获第一个错误:", err)
    }
}