Go 缓存击穿防护：singleflight 核心原理与压测实战

go get -u golang.org/x/sync

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

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

type ProductService struct {
	sf singleflight.Group
}

func (s *ProductService) GetProductInfo(ctx context.Context, id int64) (string, error) {
	cacheKey := fmt.Sprintf("product:%d", id)

	// 1. 先查 Redis (伪代码)
	// val, err := redis.Get(ctx, cacheKey)
	// if err == nil { return val, nil }

	// 2. 缓存失效，走 singleflight 合并 DB 查询
	v, err, shared := s.sf.Do(cacheKey, func() (interface{}, error) {
		// 模拟耗时的 DB 查询
		time.Sleep(50 * time.Millisecond)
		fmt.Printf(">>> 命中 DB 查询，ID: %d\n", id)
		return fmt.Sprintf("product-data-%d", id), nil
	})

	if err != nil {
		return "", err
	}

	if shared {
		// 证明有其他 Goroutine 共享了这个结果
	}

	return v.(string), nil
}

func main() {
	svc := &ProductService{}
	var wg sync.WaitGroup

	// 并发 10 个请求拿同一个 Key
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			data, _ := svc.GetProductInfo(context.Background(), 1001)
			fmt.Printf("Worker %d 拿到结果: %s\n", idx, data)
		}(i)
	}

	wg.Wait()
}

package main

import (
	"strconv"
	"testing"
	"time"

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

func mockQueryDB(id int) string {
	time.Sleep(10 * time.Millisecond)
	return "result-" + strconv.Itoa(id)
}

func BenchmarkDirectDB(b *testing.B) {
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			_ = mockQueryDB(1)
		}
	})
}

func BenchmarkSingleFlight(b *testing.B) {
	var sf singleflight.Group
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			_, _, _ = sf.Do("key:1", func() (interface{}, error) {
				return mockQueryDB(1), nil
			})
		}
	})
}

goos: darwin
goarch: arm64
pkg: bench_demo
BenchmarkDirectDB-8                100      10148520 ns/op         112 B/op          3 allocs/op
BenchmarkSingleFlight-8         118478         10086 ns/op         104 B/op          2 allocs/op
PASS
ok      bench_demo      3.421s

func (s *ProductService) BadGet(ctx context.Context, id int64) (string, error) {
    v, err, _ := s.sf.Do(fmt.Sprintf("key:%d", id), func() (interface{}, error) {
        // 错误：直接把外层的 ctx 传给了实际执行的函数
        return db.QueryWithContext(ctx, id) 
    })
    return v.(string), err
}

func (s *ProductService) SafeGet(ctx context.Context, id int64) (string, error) {
	key := fmt.Sprintf("key:%d", id)
	ch := s.sf.DoChan(key, func() (interface{}, error) {
		// 内部使用独立的、不带外层超时取消的 Context
		bgCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		return dbQuery(bgCtx, id)
	})

	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case res := <-ch:
		if res.Err != nil {
			return "", res.Err
		}
		return res.Val.(string), nil
	}
}

s.sf.Forget(cacheKey)