singleflight

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

package singleflight

import "sync"

type call struct {
    wg  sync.WaitGroup
    val interface{}
    err error
}

type Group struct {
    mu sync.Mutex     
    m  map[uint64]*call // 懒加载
}

func (g *Group) Do(key uint64, fn func() (interface{}, error)) (interface{}, error) {
  +----------------------------------+
  | g.mu.Lock()                      |
  | if g.m == nil {                  |
  |     g.m = make(map[uint64]*call) |
  | }                                |         //1号访问者访问时
  | if c, ok := g.m[key]; ok {       |         //无法从map里获取值（缓存击穿）
  |     g.mu.Unlock()                +------>  //它新建一个map的值（是个指针）
  |     c.wg.Wait()                  |         //其他访问者可以获取到该指针值，并阻塞
  |     return c.val, c.err          |         //直到1号访问者获取值后共享结果
  | }                                |         
  | c := new(call)                   |
  | c.wg.Add(1)                      |
  | g.m[key] = c                     |
  | g.mu.Unlock()                    |
  +----------------------------------+

  +----------------------+
  | c.val, c.err = fn()  |        //仅1号访问者访问，获取一次结果
  | c.wg.Done()          +----->  //结果存map里
  +----------------------+        //其他访问者复用此结果

  +------------------+
  | g.mu.Lock()      |       //仅一号访问者能运行到这
  | delete(g.m, key) +---->  //等其他请求访问完后
  | g.mu.Unlock()    |       //删除掉map中的数据，因为已经写到redis了，不需要了
  +------------------+

    return c.val, c.err
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

// server/main.go
package main

import (
	"errors"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/julienschmidt/httprouter"
)
var group *Group
var mux sync.RWMutex
var rwMux sync.RWMutex
var redisDataBase map[string]string
var mysqlDataBase map[string]string
var countRedisHit int
var countMysqlHit int

func GetFromRedis(key string) (string, error) {
	if data, ok := redisDataBase[key]; ok {
		mux.Lock()
		countRedisHit++
		mux.Unlock()
		return data, nil
	}
	if data, err := GetFromMySql(key); err == nil {
		return data, nil
	} else {
		return "", err
	}
}

func GetFromMySql(key string) (string, error) {
	time.Sleep(time.Second * 1)
	// defer fmt.Println(time.Now())
	if data, ok := mysqlDataBase[key]; ok {
		// write to redis
		rwMux.Lock()
		redisDataBase[key] = "data stored in redis"
		rwMux.Unlock()
		// 相当于设置过期时间 2s
		// 每次从存到redis缓存的值都会在2s以后过期
		go func(key string) {
			time.Sleep(time.Second * 2)
			rwMux.Lock()
			delete(redisDataBase, key)
			rwMux.Unlock()
		}(key)
		mux.Lock()
		countMysqlHit++
		mux.Unlock()
		return data, nil
	} else {
		return "", errors.New("data not found!")
	}
}

// 正常版
func GetUserInfo(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
	// 从缓存中查结果
	queryValues := req.URL.Query()
	if res, err := GetFromRedis(queryValues.Get("name")); err != nil {
		log.Fatal("err:", err)
	} else {
		fmt.Fprintf(w, res)
	}
}

// singflight版本
func GetUserInfo1(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
	// 从缓存中查结果
	queryValues := req.URL.Query()
	function := func() (interface{}, error) {
		if res, err := GetFromRedis(queryValues.Get("name")); err == nil {
			return res, nil
		} else {
			return "", err
		}
	}
	// 原来是没有共用一个group啊，我说怎么没效果
	res, err := group.Do(uint64(1), function)
	if err != nil {
		log.Fatal("err from singleflight:", err)
	}
	fmt.Fprintf(w, res.(string))
}

func ShowHitCount(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
	fmt.Fprintf(w, "redis:%d\nmysql:%d\n", countRedisHit, countMysqlHit)
}

func init() {
	redisDataBase, mysqlDataBase = map[string]string{}, map[string]string{}
	mysqlDataBase["zqw"] = "data stored in mysql" // 初始时redis里无数据
	group = &Group{}
}

func hello(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
	queryValues := req.URL.Query()
    fmt.Fprintf(w, "hello, %s!\n", queryValues.Get("name"))
}

func main() {

	router := httprouter.New()
	router.GET("/", hello)
	router.GET("/userinfo", GetUserInfo)
	router.GET("/userinfo1", GetUserInfo1)
	router.GET("/count", ShowHitCount)
	log.Fatal(http.ListenAndServe(":9090", router))

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

package main

import (
	"fmt"
	"io"
	"log"
	"net/http"
	"sync"
	"time"
)

func curl(str string) string {
	resp, err := client.Get(str)
	if err != nil {
		log.Println("error:", err)
		
	}
	res, err := io.ReadAll(resp.Body)
	defer resp.Body.Close()
	return string(res)
}

var client http.Client
func init() {
	client = http.Client{}
}
func main() {
	timeStampA := time.Now()
	defer client.CloseIdleConnections()
	var wg sync.WaitGroup
	for i := 0; i < 5000; i++ {
		// time.Sleep(time.Microsecond*1)
		wg.Add(1)
		go func() {
			defer wg.Done()
            // 下面两个轮着来
            //curl("http://localhost:9090/userinfo?name=zqw")
			curl("http://localhost:9090/userinfo1?name=zqw")
		}()
	}
	wg.Wait()
	res := curl("http://localhost:9090/count")
	fmt.Println(res)
	timeStampB := time.Now()
	fmt.Println("运行共用时: ", timeStampB.Sub(timeStampA).Seconds())
}

1
2
3
4

redis:0
mysql:5000

运行共用时:  1.5597763709999999

1
2
3
4

redis:0
mysql:1  # 其余4999次访问全部取的是第一次访问的结果

运行共用时:  1.216015732