Go语言中singleflight高并发下合并请求的最佳实践

go get golang.org/x/sync/singleflight

var sf singleflight.Group

val, err, shared := sf.Do("user:1001", func() (interface{}, error) {
    return db.QueryUser("1001") // 只有第一个请求会真正查 DB
})

fmt.Println(val, err, shared) // shared=true 表示结果是共享的

package main

import (
	"fmt"
	"net/http"
	"time"

	"github.com/gin-gonic/gin"
	"golang.org/x/sync/singleflight"
)

var (
	sf    singleflight.Group
	cache = make(map[string]string) // 简化演示，生产请用 Redis
)

func getUserHandler(c *gin.Context) {
	userID := c.Param("id")

	// 1. 先查缓存
	if val, ok := cache[userID]; ok {
		c.JSON(http.StatusOK, gin.H{"source": "cache", "data": val})
		return
	}

	// 2. 缓存未命中，使用 singleflight 合并并发请求
	// ⚠️ 注意：这里用的是 Do，同步阻塞
	val, err, shared := sf.Do("user:"+userID, func() (interface{}, error) {
		fmt.Printf("[DB] 真正查询数据库: user=%s\n", userID)
		time.Sleep(2 * time.Second) // 模拟慢查询
		return "User_" + userID, nil
	})

	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
		return
	}

	valStr := val.(string)
	cache[userID] = valStr

	c.JSON(http.StatusOK, gin.H{
		"source": "db_singleflight",
		"data":   valStr,
		"shared": shared,
	})
}

func main() {
	r := gin.Default()
	r.GET("/user/:id", getUserHandler)
	r.Run(":8080")
}

wrk -t4 -c100 -d5s http://localhost:8080/user/1001

func getUserHandler(c *gin.Context) {
	userID := c.Param("id")
	ctx := c.Request.Context() // ✅ 获取请求级别的 context

	// 1. 先查缓存（同上，省略）

	// 2. 使用 DoChan 替代 Do
	ch := sf.DoChan("user:"+userID, func() (interface{}, error) {
		fmt.Printf("[DB] 真正查询数据库: user=%s\n", userID)
		time.Sleep(2 * time.Second)
		return "User_" + userID, nil
	})

	// 3. 用 select 实现超时控制
	select {
	case result := <-ch:
		if result.Err != nil {
			// ✅ 出错时用 ForgetUnshared 清除 key，允许后续请求重试
			sf.ForgetUnshared("user:" + userID)
			c.JSON(http.StatusInternalServerError, gin.H{"error": result.Err.Error()})
			return
		}

		val := result.Val.(string)
		cache[userID] = val

		c.JSON(http.StatusOK, gin.H{
			"source": "db_singleflight",
			"data":   val,
			"shared": result.Shared,
		})
	case <-ctx.Done():
		// ✅ 客户端超时断开，服务端不再傻等
		c.JSON(http.StatusGatewayTimeout, gin.H{"error": "request timeout"})
	}
}