从 CSP 到 G-M-P：Go 并发全攻略

go f()                        // 调用函数

go func(x int){ ... }(42)     // 调用匿名函数（可捕获或显式传参）

var wg sync.WaitGroup
for i := 0; i < 3; i++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        fmt.Printf("Worker %d completed\n", id)
    }(i)
}
wg.Wait() // Join 点：等待所有分支完成

// 错误示例：所有 goroutine 打印的可能是同一个值
for i := 0; i < 5; i++ {
    go func() {
        fmt.Println(i) // 捕获的是 i 的引用
    }()
}

// 正确做法：将值作为参数传递，创建副本
for i := 0; i < 5; i++ {
    go func(j int) {
        fmt.Println(j)
    }(i) // 每次循环都传入 i 的副本
}

// producer 只能向 channel 发送数据
func producer(out chan<- int) {
    for i := 0; i < 5; i++ {
        out <- i
    }
    close(out)
}

// consumer 只能从 channel 接收数据
func consumer(in <-chan int) {
    for num := range in {
        fmt.Println("Received:", num)
    }
}

func main() {
    ch := make(chan int)

    go producer(ch) // 隐式转换为 chan<- int
    consumer(ch)    // 隐式转换为 <-chan int
}

// 有缓冲 channel 示例
ch := make(chan int, 2)
ch <- 1    // 不阻塞
ch <- 2    // 不阻塞
ch <- 3    // 阻塞，缓冲区已满

// 死锁示例
func deadlockExample() {
    ch := make(chan int)
    ch <- 1 // 死锁：无缓冲 channel，无接收者
}

ch := make(chan int)
go func() {
    // 结束时关闭，通知消费者“不会再有值”
    defer close(ch)
    for i := 0; i < 3; i++ {
        ch <- i
    }
}()

// ch 关闭后 range 自动退出
for v := range ch {
    fmt.Println(v)
}

// 所有权模式实现
func generator(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out) // 所有者负责关闭
        for _, n := range nums {
            out <- n
        }
    }()
    return out // 返回只读 channel
}

func main() {
    // 消费者只能接收，无法关闭
    for v := range generator(1, 2, 3, 4, 5) {
        fmt.Println(v)
    }
}

select {
case v := <-ch:
    // 处理数据
case <-time.After(500 * time.Millisecond):
    fmt.Println("timeout")
}

select {
case v := <-ch:
    // 处理数据
default:
    // 立即执行，不等待
}

type Context interface {
    Done() <-chan struct{}                   // 返回取消信号 Channel
    Err() error                              // 返回取消原因
    Deadline() (deadline time.Time, ok bool) // 返回截止时间
    Value(key any) any                       // 返回关联的键值对数据
}

func cancelExample() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel() // 调用 cancel 释放资源

	go func() {
		time.Sleep(2 * time.Second)
		cancel() // 模拟 2 秒后触发主动取消
	}()

	select {
	case <-time.After(5 * time.Second):
		fmt.Println("Work completed") //  5 秒后触发正常退出
	case <-ctx.Done():
		fmt.Printf("Cancelled: %v\n", ctx.Err())
	}
}

func timeoutExample() {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	select {
	case <-longRunningTask():
		fmt.Println("Task completed")
	case <-ctx.Done():
		fmt.Printf("Timeout: %v\n", ctx.Err())
	}
}

func deadlineExample() {
	deadline := time.Now().Add(5 * time.Second)
	ctx, cancel := context.WithDeadline(context.Background(), deadline)
	defer cancel()

	select {
	case <-work():
		fmt.Println("Work done")
	case <-ctx.Done():
		fmt.Printf("Deadline exceeded: %v\n", ctx.Err())
	}
}

func cascadingCancellation() {
	// 根 Context
	rootCtx, rootCancel := context.WithCancel(context.Background())

	// 子 Context 1
	childCtx1, _ := context.WithTimeout(rootCtx, 10*time.Second)

	// 子 Context 2
	childCtx2, _ := context.WithCancel(childCtx1)

	// 当 rootCancel() 被调用时，所有子 Context 都会被取消
	go worker(childCtx1, "Worker 1")
	go worker(childCtx2, "Worker 2")

	time.Sleep(2 * time.Second)
	rootCancel() // 级联取消所有子任务
}

// 定义私有键类型，避免不同包之间发生键名冲突
type contextKey string

const userIDKey contextKey = "userID"

// 封装类型安全的访问器，隐藏键的具体实现
func WithUserID(ctx context.Context, userID string) context.Context {
    return context.WithValue(ctx, userIDKey, userID)
}

func GetUserID(ctx context.Context) (string, bool) {
    userID, ok := ctx.Value(userIDKey).(string)
    return userID, ok
}

var wg sync.WaitGroup

// 在启动 Goroutine 之前调用 Add
wg.Add(len(tasks))
for _, task := range tasks {
    go func(t Task) {
        defer wg.Done() // 确保计数正确递减
        t.Execute()
    }(task)
}
wg.Wait() // 等待所有任务完成

type Counter struct {
    mu    sync.Mutex
    value int
}

func (c *Counter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock() // 确保异常情况下也能释放锁
    c.value++
}

func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

type Cache struct {
    mu   sync.RWMutex
    data map[string]interface{}
}

func (c *Cache) Get(key string) (interface{}, bool) {
    c.mu.RLock()                    // 读锁：允许并发读取
    defer c.mu.RUnlock()
    value, exists := c.data[key]
    return value, exists
}

func (c *Cache) Set(key string, value interface{}) {
    c.mu.Lock()                     // 写锁：独占访问
    defer c.mu.Unlock()
    if c.data == nil {
        c.data = make(map[string]interface{})
    }
    c.data[key] = value
}

func (c *Cache) Delete(key string) {
    c.mu.Lock()                     // 写锁：修改操作需要独占
    defer c.mu.Unlock()
    delete(c.data, key)
}

type Queue struct {
    mu    sync.Mutex
    cond  *sync.Cond
    items []interface{}
}

func NewQueue() *Queue {
    q := &Queue{}
    q.cond = sync.NewCond(&q.mu)
    return q
}

// Put 追加元素并唤醒一个等待的获取者。
func (q *Queue) Put(item interface{}) {
    q.mu.Lock()
    defer q.mu.Unlock()

    q.items = append(q.items, item)
    q.cond.Signal() // 通知一个等待者
}

// Get 阻塞直到返回一个元素。
func (q *Queue) Get() interface{} {
    q.mu.Lock()
    defer q.mu.Unlock()

    for len(q.items) == 0 { // 必须在循环中检查条件
        q.cond.Wait() // 原子地解锁并等待
    }

    item := q.items[0]
    q.items = q.items[1:]
    return item
}

var (
    once   sync.Once
    config Config
    err    error
)

func GetConfig() (Config, error) {
    once.Do(func() {
        config, err = loadConfigFromFile("config.json")
    })
    return config, err
}

// 危险：环依赖导致死锁
var (
    onceA, onceB sync.Once
    a, b         string
)

func initA() {
    onceA.Do(func() {
        initB() // initB 内部会再次触发 initA，二次进入 onceA.Do -> 阻塞
        a = "A 初始化"
    })
}

func initB() {
    onceB.Do(func() {
        initA() // 与 initA 形成循环依赖
        b = "B 初始化"
    })

// 复用可重置的 bytes.Buffer，降低短期分配与 GC 压力
var bufPool = sync.Pool{
    New: func() interface{} { return new(bytes.Buffer) },
}

// encodePayload 将任意值编码为 JSON
func encodePayload(v any) ([]byte, error) {
    b := bufPool.Get().(*bytes.Buffer)
    b.Reset()           // 清理上次内容，避免脏数据
    defer bufPool.Put(b)

    if err := json.NewEncoder(b).Encode(v); err != nil {
        return nil, err
    }

    // 注意：复制内容后再归还 Buffer，避免调用方持有的切片指向池内可变内存
    out := make([]byte, b.Len())
    copy(out, b.Bytes())
    return out, nil
}

// 数据被隔离在生成器内部，对外只提供只读视图
func generator[T any](ctx context.Context, items ...T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        // items 数据仅在此 Goroutine 内可访问
        for _, item := range items {
            select {
            case out <- item:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out // 返回只读 Channel，确保数据流向单一
}

func worker(ctx context.Context, stringStream <-chan string) <-chan struct{} {
    terminated := make(chan struct{})
    go func() {
        defer close(terminated)
        for {
            select {
            case s := <-stringStream:
                fmt.Println(s)
            case <-ctx.Done(): // 接收到退出信号
                return
            }
        }
    }()
    return terminated
}

type Result[T any] struct {
    Data  T
    Error error
}

func processURLs(ctx context.Context, urls []string) <-chan Result[*http.Response] {
    results := make(chan Result[*http.Response])

    go func() {
        defer close(results)

        for _, url := range urls {
            resp, err := http.Get(url)
            result := Result[*http.Response]{
                Data:  resp,
                Error: err,
            }

            select {
            case results <- result:
            case <-ctx.Done():
                return
            }
        }
    }()

    return results
}

// 调用方根据业务逻辑制定错误处理策略
func handleResults(ctx context.Context, urls []string) {
    for result := range processURLs(ctx, urls) {
        if result.Error != nil {
            log.Printf("Request failed: %v", result.Error)
            continue // 或根据错误类型采取不同策略
        }

        fmt.Printf("Status: %s\n", result.Data.Status)
        result.Data.Body.Close()
    }
}

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

func processFilesStructured(ctx context.Context, files []string) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(10) // 控制并发度

    for _, file := range files {
        file := file // 避免闭包捕获问题
        g.Go(func() error {
            return processFile(ctx, file)
        })
    }

    // 返回第一个错误，其他任务自动取消
    return g.Wait()
}

func worker(ctx context.Context) {
    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            fmt.Println("Worker stopping")
            return
        case <-ticker.C:
            fmt.Println("Working...")
            doWork()
        }
    }
}

func orDone[T any](ctx context.Context, in <-chan T) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		for {
			select {
			case <-ctx.Done():
				return // context 取消，关闭输出 channel
			case v, ok := <-in:
				if !ok {
					return // 输入 channel 关闭，关闭输出 channel
				}

				// 尝试发送数据，同时监听取消信号
				select {
				case out <- v:
					// 数据发送成功
				case <-ctx.Done():
					return // 在发送过程中收到取消信号
				}
			}
		}
	}()
	return out
}

// 使用示例
func processWithCancellation(ctx context.Context, input <-chan int) {
	for value := range orDone(ctx, input) {
		fmt.Println("处理:", value)
		// 无需在 for-select 循环内手动检查 ctx.Done()
	}
}

func tee[T any](ctx context.Context, in <-chan T) (<-chan T, <-chan T) {
    o1 := make(chan T)
    o2 := make(chan T)
    go func() {
        defer close(o1)
        defer close(o2)
        for v := range orDone(ctx, in) { // 使用 orDone 简化
            c1, c2 := o1, o2
            for i := 0; i < 2; i++ {
                select {
                case <-ctx.Done():
                    return
                case c1 <- v:
                    c1 = nil // 设置为 nil，防止重复写入
                case c2 <- v:
                    c2 = nil // 设置为 nil，防止重复写入
                }
            }
        }
    }()
    return o1, o2
}

func bridge[T any](ctx context.Context, streams <-chan <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for ch := range orDone(ctx, streams) {
            for v := range orDone(ctx, ch) {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}

// 泛型生成器：将一系列值转换为一个 channel 流
func generator[T any](ctx context.Context, values ...T) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		for _, v := range values {
			select {
			case <-ctx.Done():
				return
			case out <- v:
			}
		}
	}()
	return out
}

// 泛型处理阶段：对流中每个元素执行操作
func transform[T, U any](ctx context.Context, in <-chan T, fn func(T) U) <-chan U {
	out := make(chan U)
	go func() {
		defer close(out)
		for v := range in {
			select {
			case <-ctx.Done():
				return
			case out <- fn(v):
			}
		}
	}()
	return out
}

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

	intStream := generator(ctx, 1, 2, 3, 4)
	stringStream := transform(ctx, intStream, func(i int) string {
		return fmt.Sprintf("Value: %d", i*2)
	})

	for s := range stringStream {
		fmt.Println(s)
	}
}

func fanIn[T any](ctx context.Context, channels ...<-chan T) <-chan T {
    var wg sync.WaitGroup
    out := make(chan T)

    // 为每个输入启动转发 Goroutine
    for _, c := range channels {
        wg.Add(1)
        go func(ch <-chan T) {
            defer wg.Done()
            for v := range orDone(ctx, ch) {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }(c)
    }

    // 所有输入处理完后关闭输出
    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// 并行处理示例
func parallelProcess(ctx context.Context, input <-chan Task, numWorkers int) <-chan Result {
    // Fan-Out：启动多个 Worker
    workers := make([]<-chan Result, numWorkers)
    for i := 0; i < numWorkers; i++ {
        workers[i] = worker(ctx, input)
    }

    // Fan-In：聚合结果
    return fanIn(ctx, workers...)
}

func buffer[T any](ctx context.Context, upstream <-chan T, capacity int) <-chan T {
    buf := make(chan T, capacity)
    go func() {
        defer close(buf)
        for v := range orDone(ctx, upstream) {
            select {
            case <-ctx.Done():
                return
            case buf <- v:
            }
        }
    }()
    return buf
}

func workerWithHeartbeat(ctx context.Context, heartbeat chan<- struct{}) {
    for {
        select {
        case <-ctx.Done():
            return
        default:
            // 发送心跳
            select {
            case heartbeat <- struct{}{}:
            default: // 非阻塞心跳
            }
            // ... 执行工作 ...
        }
    }
}

func monitor(ctx context.Context, heartbeat <-chan struct{}, timeout time.Duration) {
    timer := time.NewTimer(timeout)
    defer timer.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-heartbeat:
            if !timer.Stop() {
                <-timer.C
            }
            timer.Reset(timeout) // 收到心跳，重置超时计时器
        case <-timer.C:
            log.Println("Worker heartbeat timeout! It might be stuck.")
            // ... 恢复/告警逻辑 ...
            return
        }
    }
}

func hedgeRequest(ctx context.Context, backends []string, query string) (string, error) {
	// 派生可取消上下文，统一控制并发请求生命周期
	gCtx, cancel := context.WithCancel(ctx)
	defer cancel() // 确保退出时通知子协程结束

	resultChan := make(chan string, 1)
	errChan := make(chan error, len(backends))

	for _, backend := range backends {
		go func(addr string) {
			// 使用 gCtx 以接收取消/超时信号
			res, err := callBackend(gCtx, addr, query)
			if err != nil {
				errChan <- err
				return
			}
			// 首个成功写入即可
			resultChan <- res
		}(backend)
	}

	// 等待：成功结果、错误，或上游取消
	select {
	case res := <-resultChan:
		return res, nil
	case err := <-errChan:
		return "", err
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

import (
	"context"
	"errors"
	"log"
	"net/http"
	"golang.org/x/time/rate"
)

// 每秒允许 2 个请求，最多允许突发 10 个请求
var limiter = rate.NewLimiter(rate.Limit(2), 10)

func handlerRequest(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	if err := limiter.Wait(ctx); err != nil {
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			http.Error(w, "Request cancelled", http.StatusRequestTimeout)
			return
		}
		http.Error(w, "Too Many Requests", http.StatusTooManyRequests)
		return
	}

	// ... 处理业务逻辑 ...
}

func supervisor[T any](
	ctx context.Context,
	workerFn func(context.Context) (<-chan T, error),
	heartbeatTimeout time.Duration
) {
	var workerCtx context.Context
	var workerCancel context.CancelFunc
	var dataStream <-chan T

	for {
		// 启动或重启 worker
		if dataStream == nil {
			workerCtx, workerCancel = context.WithCancel(ctx)
			stream, err := workerFn(workerCtx)
			if err != nil {
				log.Printf("Failed to start worker: %v. Retrying...", err)
				// 避免快速失败循环
				select {
				case <- time.After(time.Second):
				case <- ctx.Done():
					return
				}
				continue
			}
			dataStream = stream
		}

		timer := time.NewTimer(heartbeatTimeout)

		select {
		case <-ctx.Done():
			workerCancel()
			return
		case data, ok := <-dataStream:
			timer.Stop()
			if !ok { // worker 退出
				dataStream = nil // 准备在下一次循环中重启
				continue
			}
			log.Printf("Received data: %v", data)
		case <-timer.C: // 心跳超时
			log.Println("Worker heartbeat timeout. Restarting...")
			workerCancel()   //终止旧 worker
			dataStream = nil // 准备重启
		}
	}
}