介绍完github.com/golang/groupcache如何使用和基本原理后，我们来分析下它的源码。初始化的代码位于groupcache.go
代码语言：javascript复制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{},初始化了单飞，后面获取数据就是基于单飞的。
代码语言：javascript复制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的时候注册进去的。其中回源方法的定义如下：
代码语言：javascript复制type GetterFunc func(ctx context.Context, key string, dest Sink) errorfunc (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
代码语言：javascript复制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里面去查询的
代码语言：javascript复制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里面定义了时间存储数据的槽
代码语言：javascript复制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获取
代码语言：javascript复制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结构里
代码语言：javascript复制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值即可。
代码语言：javascript复制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
}
