高级并发模式深入

// fanOut 把 in 分发给 workers 个 goroutine，各自返回一个输出 channel。
func fanOut(ctx context.Context, in <-chan int, workers int) []<-chan int {
	chans := make([]<-chan int, workers)
	for i := range chans {
		out := make(chan int) // 无缓冲：天然背压，下游不读上游就停
		chans[i] = out
		go func() {
			defer close(out) // 只关自己创建的 channel
			for v := range in {
				select {
				case out <- v * 2: // 并行处理
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	return chans
}

// fanIn 把多个 channel 合并成一个。
func fanIn(ctx context.Context, chans ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	for _, ch := range chans {
		wg.Go(func() { // WaitGroup.Go：Go 1.25+，等价于 Add(1)+go+Done
			for v := range ch {
				select {
				case out <- v:
				case <-ctx.Done():
					return
				}
			}
		})
	}
	// 关闭必须放在独立 goroutine：否则主流程会卡在这里，一个值都读不到
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

// 生产者：每一跳发送都带 ctx.Done
func source(ctx context.Context, nums []int) <-chan int {
	in := make(chan int)
	go func() {
		defer close(in)
		for _, n := range nums {
			select {
			case in <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return in
}

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

results := fanIn(ctx, fanOut(ctx, source(ctx, nums), 5)...)
for v := range results {
    if v > 100 {
        break // 提前退出 → defer 的 cancel() 兜底，整条链一起收
    }
}

func generator(ctx context.Context, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case out <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case out <- n * n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func filter(ctx context.Context, in <-chan int, keep func(int) bool) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			if keep(n) {
				select {
				case out <- n:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out
}

// 组装：1..10 → 平方 → 只留 >50
result := filter(ctx, square(ctx, generator(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)),
	func(n int) bool { return n > 50 })
for v := range result {
	fmt.Println(v) // 64, 81, 100
}

type Result[T any] struct {
	Value T
	Err   error
}

func safeSquare(ctx context.Context, in <-chan int) <-chan Result[int] {
	out := make(chan Result[int])
	go func() {
		defer close(out)
		for n := range in {
			if n < 0 {
				select {
				case out <- Result[int]{Err: fmt.Errorf("square: negative input %d", n)}:
				case <-ctx.Done():
					return
				}
				continue
			}
			select {
			case out <- Result[int]{Value: n * n}:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func gen(nums ...int) iter.Seq[int] {
	return func(yield func(int) bool) {
		for _, n := range nums {
			if !yield(n) {
				return // 下游 break，这里立刻收到 false
			}
		}
	}
}

func squared(in iter.Seq[int]) iter.Seq[int] {
	return func(yield func(int) bool) {
		for n := range in {
			if !yield(n * n) {
				return
			}
		}
	}
}

for v := range squared(gen(1, 2, 3, 4)) {
	if v == 9 {
		break // 上游 stage 全部自动停止，没有 goroutine 泄漏，也不需要 ctx
	}
}

type Job struct {
	ID      int
	Payload string
}

type Result struct {
	JobID int
	Data  string
	Err   error
}

type WorkerPool struct {
	jobs      chan Job
	results   chan Result
	workers   int
	wg        sync.WaitGroup
	closeOnce sync.Once
}

func NewWorkerPool(workers, queueSize int) *WorkerPool {
	return &WorkerPool{
		jobs:    make(chan Job, queueSize),
		results: make(chan Result, queueSize),
		workers: workers,
	}
}

func (p *WorkerPool) Start(ctx context.Context) {
	for range p.workers {
		p.wg.Go(func() { p.run(ctx) })
	}
	go func() {
		p.wg.Wait()
		close(p.results) // 唯一的关闭点：所有 worker 都退出之后
	}()
}

func (p *WorkerPool) run(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case job, ok := <-p.jobs:
			if !ok {
				return // jobs 被关闭：正常收尾
			}
			res := Result{JobID: job.ID, Data: strings.ToUpper(job.Payload)}
			// 关键：往 results 发也要能被取消，否则 ctx 已死、无人消费时这里永久阻塞
			select {
			case p.results <- res:
			case <-ctx.Done():
				return
			}
		}
	}
}

// Submit 带 ctx：队列满时调用方可被取消，而不是傻等
func (p *WorkerPool) Submit(ctx context.Context, job Job) error {
	select {
	case p.jobs <- job:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

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

// Close 幂等：只关 jobs，绝不关 results（那由 Start 里的收尾 goroutine 负责）
func (p *WorkerPool) Close() { p.closeOnce.Do(func() { close(p.jobs) }) }

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(8) // 并发上限，等价于 8 个 worker
for _, job := range jobs {
    g.Go(func() error { return process(ctx, job) })
}
if err := g.Wait(); err != nil {
    // 第一个错误；ctx 已被自动取消，其余任务尽快退出
}

type Future[T any] struct {
	result T
	err    error
	done   chan struct{} // 只用来广播"完成"，不传数据
}

func Start[T any](fn func() (T, error)) *Future[T] {
	f := &Future[T]{done: make(chan struct{})}
	go func() {
		f.result, f.err = fn()
		close(f.done) // 广播：所有等待者同时被唤醒
	}()
	return f
}

func (f *Future[T]) Get() (T, error) {
	<-f.done
	return f.result, f.err
}

func (f *Future[T]) GetWithin(d time.Duration) (T, error) {
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case <-f.done:
		return f.result, f.err
	case <-timer.C:
		var zero T
		return zero, fmt.Errorf("future: timeout after %v", d)
	}
}

userF := Start(func() (string, error) { return fetchUser(123) })
orderF := Start(func() ([]Order, error) { return fetchOrders(123) })

user, err := userF.Get()   // 三个 RPC 同时在飞，总耗时 ≈ 最慢的那个
orders, err := orderF.Get()

func asyncCompute(ctx context.Context, input int) <-chan Result {
	ch := make(chan Result, 1) // 缓冲 1：无人接收时 goroutine 也能退出
	go func() {
		v, err := expensiveComputation(ctx, input)
		ch <- Result{Value: v, Err: err} // 有缓冲，永不阻塞
		close(ch)
	}()
	return ch
}

var getConf = sync.OnceValue(func() *Config { return loadConfig() }) // 惰性、只算一次、所有人等它

type RateLimiter struct {
	tokens chan struct{}
	ticker *time.Ticker
	stop   context.CancelFunc
	once   sync.Once
}

func NewRateLimiter(ratePerSec, burst int) (*RateLimiter, error) {
	if ratePerSec <= 0 || burst <= 0 {
		return nil, errors.New("rate limiter: rate and burst must be positive")
	}
	interval := time.Second / time.Duration(ratePerSec)
	if interval <= 0 {
		return nil, fmt.Errorf("rate limiter: rate %d too high for ticker interval", ratePerSec)
	}
	ctx, cancel := context.WithCancel(context.Background())
	rl := &RateLimiter{
		tokens: make(chan struct{}, burst),
		ticker: time.NewTicker(interval),
		stop:   cancel,
	}
	for range burst {
		rl.tokens <- struct{}{} // 预填满，允许突发
	}
	go func() {
		defer rl.ticker.Stop()
		for {
			select {
			case <-rl.ticker.C:
				select {
				case rl.tokens <- struct{}{}:
				default: // 桶满：丢弃令牌，不排队补偿
				}
			case <-ctx.Done():
				return
			}
		}
	}()
	return rl, nil
}

func (rl *RateLimiter) Allow() bool { // 非阻塞：拿不到令牌直接失败
	select {
	case <-rl.tokens:
		return true
	default:
		return false
	}
}

func (rl *RateLimiter) Wait(ctx context.Context) error { // 阻塞：等到令牌或被取消
	select {
	case <-rl.tokens:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (rl *RateLimiter) Close() { rl.once.Do(rl.stop) }

// 经典死锁：两个 goroutine 以相反顺序拿两把锁
var mu1, mu2 sync.Mutex

go func() {
	mu1.Lock()
	defer mu1.Unlock()
	time.Sleep(time.Millisecond)
	mu2.Lock() // 等 mu2
	defer mu2.Unlock()
}()

go func() {
	mu2.Lock() // 与上面相反
	defer mu2.Unlock()
	time.Sleep(time.Millisecond)
	mu1.Lock() // 等 mu1 → 互等
	defer mu1.Unlock()
}()

type Transfer struct {
	mu       sync.Mutex
	balance  map[string]int
}

type Account struct {
	ID  string
	mu  sync.Mutex
	bal int
}

// 所有调用者都按 ID 升序拿锁 → 不可能出现环路等待
func transfer(from, to *Account, amount int) {
	a, b := from, to
	if a.ID > b.ID {
		a, b = b, a
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	b.mu.Lock()
	defer b.mu.Unlock()
	if a.bal < amount {
		return
	}
	a.bal -= amount
	b.bal += amount
}

func lockBoth(mu1, mu2 *sync.Mutex) bool {
	if !mu1.TryLock() {
		return false
	}
	if !mu2.TryLock() {
		mu1.Unlock() // 只解锁自己已经持有的那把
		return false
	}
	return true
}

if !lockBoth(&a.mu, &b.mu) {
	time.Sleep(time.Millisecond) // 退避
	// 重试
} else {
	defer a.mu.Unlock()
	defer b.mu.Unlock()
	// ...
}

// 形态 1：无人接收的发送
func leakSend() {
	ch := make(chan int)      // 无缓冲
	go func() { ch <- 42 }()  // 永久阻塞在这里
	// ch 被丢弃，再没人读
}

// 形态 2：生产者忽略取消信号（第 1 节实测：cancel 后仍剩 2 个 goroutine）
func leakProducer(ctx context.Context, out chan<- int) {
	go func() {
		for i := range 100000 {
			out <- i // 只有一行发送，没有 select ctx.Done()
		}
	}()
}

// 形态 3：select 里没有退出分支
func leakLoop(ctx context.Context) {
	go func() {
		for {
			select {
			case v := <-workCh:
				handle(v)
				// 忘了 case <-ctx.Done()：ctx 取消也停不下来
			}
		}
	}()
}

// 形态 4：等一个永远不会到来的 close
func leakRange() {
	ch := make(chan int)
	go func() {
		for v := range ch { // 发送方忘了 close → 永久 range
			use(v)
		}
	}()
}

// 形态 1：给一次性结果加缓冲 1（或让接收方一定存在）
func fixedSend() {
	ch := make(chan int, 1)
	go func() { ch <- 42 }() // 立刻完成，goroutine 退出
}

// 形态 2/3：每一跳发送/等待都带 ctx.Done 分支
func fixedLoop(ctx context.Context) {
	timer := time.NewTimer(time.Second)
	defer timer.Stop()
	for {
		select {
		case <-timer.C:
			doWork()
			timer.Reset(time.Second) // Go 1.23+ 直接 Reset，不必先排空
		case <-ctx.Done():
			return
		}
	}
}

n := runtime.NumGoroutine() // 打点：压测前后、每个发布版本的基线

// 方案 1（首选）：一个协调 goroutine + WaitGroup 等所有人写完再关
var wg sync.WaitGroup
for _, item := range items {
	wg.Go(func() { produce(jobsCh, item) }) // 只写，不关
}
go func() {
	wg.Wait()
	close(jobsCh) // 唯一的关闭者
}()

// 方案 2：用 done channel 由所有者下命令，生产者退出，关闭仍在所有者手里
// 方案 3：根本不关，接收方靠 ctx.Done 或哨兵值结束（语义模糊，不推荐）

go test -race ./...
go run -race main.go
go build -race -o myapp
go install -race mypkg