-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathgroup.go
136 lines (119 loc) · 2.88 KB
/
group.go
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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
package cache
import (
"fmt"
"log"
pb "mini-cache/cachepb"
"mini-cache/singleflight"
"sync"
)
// A Getter loads data for a key.
type Getter interface {
Get(key string) ([]byte, error)
}
// A GetterFunc implements Getter with a function.
type GetterFunc func(key string) ([]byte, error)
// Get implements Getter interface function.
func (f GetterFunc) Get(key string) ([]byte, error) {
return f(key)
}
// A Group is a cache namespace and associated data loaded spread over
type Group struct {
// 缓存类型命名
name string
// 缓存未命中时获取源数据回调
getter Getter
// 并发缓存
mainCache cache
peers PeerPicker
// use singleflight.Group to make sure that
// each key is only fetched once.
loader *singleflight.Group
}
var (
mu sync.RWMutex
groups = make(map[string]*Group)
)
// 实例化 Group,并存储在全局 groups map
func NewGroup(name string, cacheBytes int64, getter Getter) *Group {
if getter == nil {
panic("nil Getter")
}
mu.Lock()
defer mu.Unlock()
g := &Group{
name: name,
getter: getter,
mainCache: cache{cacheBytes: cacheBytes},
loader: &singleflight.Group{},
}
groups[name] = g
return g
}
func GetGroup(name string) *Group {
// 不涉及写操作,使用只读锁
mu.RLock()
g := groups[name]
mu.RUnlock()
return g
}
// 核心方法,判断是否命中
func (g *Group) Get(key string) (ByteView, error) {
if key == "" {
return ByteView{}, fmt.Errorf("key is required")
}
if v, ok := g.mainCache.get(key); ok {
log.Println("[MiniCache] hit")
return v, nil
}
return g.load(key)
}
func (g *Group) load(key string) (value ByteView, err error) {
viewi, err := g.loader.Do(key, func() (interface{}, error) {
if g.peers != nil {
if peer, ok := g.peers.PickPeer(key); ok {
if value, err = g.getFromPeer(peer, key); err == nil {
return value, nil
}
log.Println("[MiniCache] Failed to get from peer", err)
}
}
return g.getLocally(key)
})
if err == nil {
return viewi.(ByteView), nil
}
return
}
func (g *Group) getLocally(key string) (ByteView, error) {
bytes, err := g.getter.Get(key)
if err != nil {
return ByteView{}, err
}
value := ByteView{b: cloneBytes(bytes)}
g.populateCache(key, value)
return value, nil
}
// 添加到缓存
func (g *Group) populateCache(key string, value ByteView) {
g.mainCache.add(key, value)
}
// 将实现了 PeerPicker 接口的 HTTPPool 注入 Group
func (g *Group) RegisterPeers(peers PeerPicker) {
if g.peers != nil {
panic("RegisterPeerPicker called more than once")
}
g.peers = peers
}
// 使用实现了 PeerGetter 接口的 httpGetter 从访问远程节点获取缓存值
func (g *Group) getFromPeer(peer PeerGetter, key string) (ByteView, error) {
req := &pb.Request{
Group: g.name,
Key: key,
}
res := &pb.Response{}
err := peer.Get(req, res)
if err != nil {
return ByteView{}, err
}
return ByteView{b: res.Value}, nil
}