golang 源碼分析：groupcache（2）

介紹完 github.com/golang/groupcache 如何使用和基本原理後，我們來分析下它的源碼。初始化的代碼位於 groupcache.go

func NewGroup(name string, cacheBytes int64, getter Getter) *Group { 
return newGroup(name, cacheBytes, getter, nil)
}

func newGroup(name string, cacheBytes int64, getter Getter, peers PeerPicker) *Group {
initPeerServerOnce.Do(callInitPeerServer)
g := &Group{
name:       name,
getter:     getter,
peers:      peers,
cacheBytes: cacheBytes,
loadGroup:  &singleflight.Group{},
}
if fn := newGroupHook; fn != nil {
fn(g)
}
groups[name] = g
return g

重點注意下其中 loadGroup: &singleflight.Group{}, 初始化了單飛，後面獲取數據就是基於單飛的。

var newGroupHook func(*Group)

type Group struct {
name       string
getter     Getter
peersOnce  sync.Once
peers      PeerPicker
cacheBytes int64 // limit for sum of mainCache and hotCache size
// mainCache is a cache of the keys for which this process
// (amongst its peers) is authoritative. That is, this cache
// contains keys which consistent hash on to this process's
// peer number.
mainCache cache
// hotCache contains keys/values for which this peer is not
// authoritative (otherwise they would be in mainCache), but
// are popular enough to warrant mirroring in this process to
// avoid going over the network to fetch from a peer.  Having
// a hotCache avoids network hotspotting, where a peer's
// network card could become the bottleneck on a popular key.
// This cache is used sparingly to maximize the total number
// of key/value pairs that can be stored globally.
hotCache cache
// loadGroup ensures that each key is only fetched once
// (either locally or remotely), regardless of the number of
// concurrent callers.
loadGroup flightGroup
_ int32 // force Stats to be 8-byte aligned on 32-bit platforms
// Stats are statistics on the group.
Stats Stats
}

func callInitPeerServer() {
if initPeerServer != nil {
initPeerServer()
}
}

func RegisterServerStart(fn func()) {
if initPeerServer != nil {
panic("RegisterServerStart called more than once")
}
initPeerServer = fn
}

其中 callInitPeerServer 方法是我們在初始化 server 的時候註冊進去的。其中回源方法的定義如下：

type GetterFunc func(ctx context.Context, key string, dest Sink) error

func (g *Group) Get(ctx context.Context, key string, dest Sink) error {
g.peersOnce.Do(g.initPeers)
value, cacheHit := g.lookupCache(key)
value, destPopulated, err := g.load(ctx, key, dest)

我們查詢的時候分爲三步，initpeer，從本地查詢，到 peer 查詢或者回源。本地 cache 包括 maincache 和 hotcache

func (g *Group) lookupCache(key string) (value ByteView, ok bool) {
if g.cacheBytes <= 0 {
return
}
value, ok = g.mainCache.get(key)
if ok {
return
}
value, ok = g.hotCache.get(key)
return
}

func (g *Group) initPeers() {
if g.peers == nil {
g.peers = getPeers(g.name)
}
}

具體查詢，是到 lru 裏面去查詢的

func (c *cache) get(key string) (value ByteView, ok bool) {
c.mu.Lock()
defer c.mu.Unlock()
c.nget++
if c.lru == nil {
return
}
vi, ok := c.lru.Get(key)
if !ok {
return
}
c.nhit++
return vi.(ByteView), true
}

type flightGroup interface {
// Done is called when Do is done.
Do(key string, fn func() (interface{}, error)) (interface{}, error)
}

func (g *Group) load(ctx context.Context, key string, dest Sink) (value ByteView, destPopulated bool, err error) {
g.Stats.Loads.Add(1)
viewi, err := g.loadGroup.Do(key, func() (interface{}, error) {
if value, cacheHit := g.lookupCache(key); cacheHit {
if peer, ok := g.peers.PickPeer(key); ok {
value, err = g.getFromPeer(ctx, peer, key)
value, err = g.getLocally(ctx, key, dest)
g.populateCache(key, value, &g.mainCache)

func (g *Group) getLocally(ctx context.Context, key string, dest Sink) (ByteView, error) {
err := g.getter.Get(ctx, key, dest)
if err != nil {
return ByteView{}, err
}
return dest.view()
}

func (g *Group) populateCache(key string, value ByteView, cache *cache) {
cache.add(key, value)
victim := &g.mainCache
victim.removeOldest()

func (c *cache) add(key string, value ByteView) {
c.mu.Lock()
defer c.mu.Unlock()
if c.lru == nil {
c.lru = &lru.Cache{
OnEvicted: func(key lru.Key, value interface{}) {
val := value.(ByteView)
c.nbytes -= int64(len(key.(string))) + int64(val.Len())
c.nevict++
},
}
}
c.lru.Add(key, value)
c.nbytes += int64(len(key)) + int64(value.Len())
}

sinks.go 裏面定義了時間存儲數據的槽

type Sink interface {
// SetString sets the value to s.
SetString(s string) error
// SetBytes sets the value to the contents of v.
// The caller retains ownership of v.
SetBytes(v []byte) error
// SetProto sets the value to the encoded version of m.
// The caller retains ownership of m.
SetProto(m proto.Message) error
// view returns a frozen view of the bytes for caching.
view() (ByteView, error)
}

func AllocatingByteSliceSink(dst *[]byte) Sink {
return &allocBytesSink{dst: dst}
}

type allocBytesSink struct {
dst *[]byte
v   ByteView
}

peers.go 裏面定義瞭如何獲取 peer，默認是採用一致性 hash 獲取

func getPeers(groupName string) PeerPicker {
if portPicker == nil {
return NoPeers{}
}
pk := portPicker(groupName)
if pk == nil {
pk = NoPeers{}
}
return pk
}

var (
portPicker func(groupName string) PeerPicker
)

type PeerPicker interface {
// PickPeer returns the peer that owns the specific key
// and true to indicate that a remote peer was nominated.
// It returns nil, false if the key owner is the current peer.
PickPeer(key string) (peer ProtoGetter, ok bool)
}

http.go 裏把 peer 存入一致性 hash 結構裏

func (p *HTTPPool) Set(peers ...string) {
p.mu.Lock()
defer p.mu.Unlock()
p.peers = consistenthash.New(p.opts.Replicas, p.opts.HashFn)
p.peers.Add(peers...)
p.httpGetters = make(map[string]*httpGetter, len(peers))
for _, peer := range peers {
p.httpGetters[peer] = &httpGetter{transport: p.Transport, baseURL: peer + p.opts.BasePath}
}
}

func NewHTTPPool(self string) *HTTPPool {
p := NewHTTPPoolOpts(self, nil)
http.Handle(p.opts.BasePath, p)
return p
}

func NewHTTPPoolOpts(self string, o *HTTPPoolOptions) *HTTPPool {
p := &HTTPPool{
self:        self,
httpGetters: make(map[string]*httpGetter),
}
p.peers = consistenthash.New(p.opts.Replicas, p.opts.HashFn)
RegisterPeerPicker(func() PeerPicker { return p })

consistenthash/consistenthash.go 定義了一致性 hash 的實現，假設副本數量是 n，它會通過 i 到 n+key 來計算出 hash 值，存入整體的 hash 值 list 裏面，然後對 hash 值排序。並且用 map 存 hash 值到 key 的映射。查找的時候同樣算法找到對應的 hash 值即可。

func New(replicas int, fn Hash) *Map {
m := &Map{
replicas: replicas,
hash:     fn,
hashMap:  make(map[int]string),
}
if m.hash == nil {
m.hash = crc32.ChecksumIEEE
}
return m
}

func (m *Map) Add(keys ...string) {
for _, key := range keys {
for i := 0; i < m.replicas; i++ {
hash := int(m.hash([]byte(strconv.Itoa(i) + key)))
m.keys = append(m.keys, hash)
m.hashMap[hash] = key
}
}
sort.Ints(m.keys)
}

func (m *Map) Get(key string) string {
if m.IsEmpty() {
return ""
}
hash := int(m.hash([]byte(key)))
// Binary search for appropriate replica.
idx := sort.Search(len(m.keys), func(i int) bool { return m.keys[i] >= hash })
// Means we have cycled back to the first replica.
if idx == len(m.keys) {
idx = 0
}
return m.hashMap[m.keys[idx]]
}

func (p *HTTPPool) PickPeer(key string) (ProtoGetter, bool) {
p.mu.Lock()
defer p.mu.Unlock()
if p.peers.IsEmpty() {
return nil, false
}
if peer := p.peers.Get(key); peer != p.self {
return p.httpGetters[peer], true
}
return nil, false
}

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/onxq4AAUXsjQctNRTjvSVg

猜你喜歡