Go 常用并发模式实战与 Benchmark 性能剖析

package main

import (
	"context"
	"sync"
)

type Task struct {
	ID int
}

func Worker(ctx context.Context, id int, tasks <-chan Task, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case task, ok := <-tasks:
			if !ok {
				return
			}
			// 处理业务逻辑
			_ = task.ID * 2
		}
	}
}

func RunWorkerPool(ctx context.Context, workerCount int, taskList []Task) {
	tasks := make(chan Task, 100)
	var wg sync.WaitGroup

	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go Worker(ctx, i, tasks, &wg)
	}

	for _, task := range taskList {
		tasks <- task
	}
	close(tasks)

	wg.Wait()
}

goos: darwin
goarch: arm64
pkg: demo/concurrency
Benchmark_UnboundedGoroutines-8       12    92410200 ns/op   24510200 B/op   100120 allocs/op
Benchmark_WorkerPool_50-8             48    24150300 ns/op     802340 B/op      124 allocs/op

func FetchAggregatedData(ctx context.Context) ([]string, error) {
	ctx, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
	defer cancel()

	results := make(chan string, 3)
	var wg sync.WaitGroup

	sources := []func(context.Context) string{
		fetchUserProfile,
		fetchOrderStats,
		fetchRiskScore,
	}

	// Fan-Out
	for _, fn := range sources {
		wg.Add(1)
		go func(f func(context.Context) string) {
			defer wg.Done()
			select {
			case <-ctx.Done():
				return
			case results <- f(ctx):
			}
		}(fn)
	}

	// 监听完成并关闭 Channel
	go func() {
		wg.Wait()
		close(results)
	}()

	// Fan-In 收集
	var finalData []string
	for res := range results {
		finalData = append(finalData, res)
	}

	if ctx.Err() != nil {
		return nil, ctx.Err()
	}
	return finalData, nil
}

// Stage 1: 生成数据
func gen(ctx context.Context, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case <-ctx.Done():
				return
			case out <- n:
			}
		}
	}()
	return out
}

// Stage 2: 处理数据 (平方)
func sq(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case <-ctx.Done():
				return
			case out <- n * n:
			}
		}
	}()
	return out
}

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

var g singleflight.Group

func GetProductDetail(productID string) (string, error) {
	// 命中本地或 Redis 缓存省略...
	
	// 缓存未命中，使用 Singleflight 合并下游 DB 请求
	v, err, shared := g.Do(productID, func() (interface{}, error) {
		// 模拟慢查询 DB
		time.Sleep(50 * time.Millisecond)
		return queryDB(productID)
	})

	if err != nil {
		return "", err
	}
	
	_ = shared // shared == true 表示该请求共享了其他人的结果
	return v.(string), nil
}