groupcache源码解读

Go 2019-11-21

Simple usage

gropucache的官方网站是 https://github.com/golang/groupcache

consistenthash模块

一致性hash算法，通常是用在查找一个合适的下载节点时，使负载更平均，但是对于同样的请求始终返回一样的结果。

type Map struct {
    hash     Hash
    replicas int
    keys     []int // Sorted
    hashMap  map[int]string
}

Map结构中replicas的含义是增加虚拟桶，使数据分布更加均匀。

// 创建Map结构
func New(replicas int,fn Hash) *Map
// 添加新的Key
func (m *Map) Add(keys ...string)
// 根据hash(key)获取value
func (m *Map) Get(key string) string
// 判断Map是否为空
func (m *Map) IsEmpty() bool
// 竟然没有提供Remove方法 O_O

用法简单例子

package main

import (
	"fmt"

	"github.com/golang/groupcache/consistenthash"
)

func main() {
	c := consistenthash.New(70,nil)
	c.Add("A","B","C","D","E")
	for _,key := range []string{"what","nice","what","good","yes!"} {
		fmt.Printf("%s -> %s\n",key,c.Get(key))
	}
}

// Expect output
// -------------
// what -> C
// nice -> A
// what -> C
// nice -> A
// good -> D
// yes! -> E

singleflight 模块

type Group struct {
}

// 当多个相同的key请求的时候，函数只被调用一次
func (g *Group) Do(key string,fn func() (interface{},error)) (interface{},error)

使用例子

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/golang/groupcache/singleflight"
)

func NewDelayReturn(dur time.Duration,n int) func() (interface{},error) {
	return func() (interface{},error) {
		time.Sleep(dur)
		return n,nil
	}
}

func main() {
	g := singleflight.Group{}
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		ret,err := g.Do("key",NewDelayReturn(time.Second*1,1))
		if err != nil {
			panic(err)
		}
		fmt.Printf("key-1 get %v\n",ret)
		wg.Done()
	}()
	go func() {
		time.Sleep(100 * time.Millisecond) // make sure this is call is later
		ret,NewDelayReturn(time.Second*2,2))
		if err != nil {
			panic(err)
		}
		fmt.Printf("key-2 get %v\n",ret)
		wg.Done()
	}()
	wg.Wait()
}

执行结果(耗时： 1.019s)

key-2 get 1
key-1 get 1

lru 模块

最初是用在内存管理上的一个算法，根据历史的请求数，分析出最热门的数据，并保存下来。

type Cache struct {
    // MaxEntries is the maximum number of cache entries before
    // an item is evicted. Zero means no limit.
    MaxEntries int

    // OnEvicted optionally specificies a callback function to be
    // executed when an entry is purged from the cache.
    OnEvicted func(key Key,value interface{})
    // contains filtered or unexported fields
}

func New(maxEntries int) *Cache
func (c *Cache) Add(key Key,value interface{})
func (c *Cache) Get(key Key) (value interface{},ok bool)
func (c *Cache) Len() int
func (c *Cache) Remove(key Key)
func (c *Cache) RemoveOldest()

用法举例

package main

import (
        "fmt"

        "github.com/golang/groupcache/lru"
)

func main() {
        cache := lru.New(2)
        cache.Add("x","x0")
        cache.Add("y","y0")
        yval,ok := cache.Get("y")
        if ok {
                fmt.Printf("y is %v\n",yval)
        }
        cache.Add("z","z0")

        fmt.Printf("cache length is %d\n",cache.Len())
        _,ok = cache.Get("x")
        if !ok {
                fmt.Printf("x key was weeded out\n")
        }
}
// Expect output
//--------------
// y is y0
// cache length is 2
// x key was weeded out

HTTPPool 模块

type HTTPPool struct {
    // 可选，为每次的请求封装的Context参数
    Context func(*http.Request) groupcache.Context

    // 可选，不懂这个干啥的
    // 注释说：请求的时候用的就是这个
    Transport func(groupcache.Context) http.RoundTripper
}

// self 必须是一个合法的URL指向当前的服务器，比如 "http://10.0.0.1:8000"
// 这个函数默会注册一个路由
// http.handle("/_groupcache/",poolInstance) 
// 该路由主要用户节点间获取数据的功能
// 另外该函数不能重复调用，否则会panic
func NewHTTPPool(self string) *HTTPPool

// 更新节点列表
// 用了consistenthash
// 奇怪的时候，只有节点添加的函数，并没有删除的
func (p *HTTPPool) Set(peers ...string)

// 用一致性hash算法选择一个节点
func (p *HTTPPool) PickPeer(key string) (groupcache.ProtoGetter,bool)

// 用于处理通过HTTP传递过来的grpc请求
func (p *HTTPPool) ServeHTTP(w http.ResponseWriter,r *http.Request)

其他 (待补充）

本来想在官网上找个例子，可以

package main

// A SizeReaderAt is a ReaderAt with a Size method.
//
// An io.SectionReader implements SizeReaderAt.
type SizeReaderAt interface {
    Size() int64
    io.ReaderAt
}

// NewMultiReaderAt is like io.MultiReader but produces a ReaderAt
// (and Size),instead of just a reader.
func NewMultiReaderAt(parts ...SizeReaderAt) SizeReaderAt {
    m := &multi{
        parts: make([]offsetAndSource,len(parts)),}
    var off int64
    for _,p := range parts {
        m.parts = append(m.parts,offsetAndSource{off,p})
        off += p.Size()
    }
    m.size = off
    return m
}

// NewChunkAlignedReaderAt returns a ReaderAt wrapper that is backed
// by a ReaderAt r of size totalSize where the wrapper guarantees that
// all ReadAt calls are aligned to chunkSize boundaries and of size
// chunkSize (except for the final chunk,which may be shorter).
//
// A chunk-aligned reader is good for caching,letting upper layers have
// any access pattern,but guarantees that the wrapped ReaderAt sees
// only nicely-cacheable access patterns & sizes.
func NewChunkAlignedReaderAt(r SizeReaderAt,chunkSize int) SizeReaderAt {
    // ...
}

func part(s string) SizeReaderAt {
    return io.NewSectionReader(strings.NewReader(s),int64(len(s)))
}

func handler(w http.ResponseWriter,r *http.Request) {
    sra := NewMultiReaderAt(
        part("Hello,"),part(" world! "),part("You requested "+r.URL.Path+"\n"),)
    rs := io.NewSectionReader(sra,sra.Size())
    http.ServeContent(w,r,"foo.txt",modTime,rs)
}


func main(){
    me := "http://10.0.0.1"
    peers := groupcache.NewHTTPPool(me)

    // Whenever peers change:
    peers.Set("http://10.0.0.1","http://10.0.0.2","http://10.0.0.3")
        
    var thumbNails = groupcache.NewGroup("thumbnail",64<<20,groupcache.GetterFunc(
        func(ctx groupcache.Context,key string,dest groupcache.Sink) error {
            fileName := key
            dest.SetBytes(generateThumbnail(fileName))
            return nil
        }))

    var data []byte
    err := thumbNails.Get(ctx,"big-file.jpg",groupcache.AllocatingByteSliceSink(&data))
    // ...
    http.ServeContent(w,"big-file-thumb.jpg",bytes.NewReader(data))


            
}

groupcache 使用例子

package main

import (
        "errors"
        "flag"
        "log"
        "net/http"
        "strconv"
        "strings"

        "github.com/golang/groupcache"
)

var localStorage map[string]string

func init() {
        localStorage = make(map[string]string)
        localStorage["hello"] = "world"
        localStorage["info"] = "This is an example"
}

func main() {
        port := flag.Int("port",4100,"Listen port")
        flag.Parse()

        // Name have to starts with http://
        self := "http://localhost:" + strconv.Itoa(*port)
        pool := groupcache.NewHTTPPool(self)
        pool.Set(self,"http://localhost:4101")

        var helloworld = groupcache.NewGroup("helloworld",10<<20,groupcache.GetterFunc(
                func(ctx groupcache.Context,dest groupcache.Sink) error {
                        log.Printf("groupcache get key: %v",key)
                        value,exists := localStorage[key]
                        if !exists {
                                dest.SetString(key + " NotExist")
                                return errors.New(key + " NotExist")
                        } else {
                                dest.SetString(value)
                                return nil
                        }
                }))

        http.HandleFunc("/",func(w http.ResponseWriter,r *http.Request) {
                key := strings.TrimPrefix(r.RequestURI,"/")
                log.Printf("Request(%v) key(%v)",r.RemoteAddr,key)
                if key == "" {
                        http.Error(w,"Bad Request",http.StatusBadRequest)
                        return
                }
                var data []byte
                err := helloworld.Get(nil,groupcache.AllocatingByteSliceSink(&data))
                if err != nil {
                        http.Error(w,err.Error(),http.StatusBadRequest)
                        return
                }
                log.Printf("cache data: %v",data)
                w.Write(data)
                log.Println("Gets: ",helloworld.Stats.Gets.String())
                log.Println("CacheHits: ",helloworld.Stats.CacheHits.String())
                log.Println("Loads: ",helloworld.Stats.Loads.String())
                log.Println("LocalLoads: ",helloworld.Stats.LocalLoads.String())
                log.Println("PeerErrors: ",helloworld.Stats.PeerErrors.String())
                log.Println("PeerLoads: ",helloworld.Stats.PeerLoads.String())
        })
        http.ListenAndServe(":"+strconv.Itoa(*port),nil)
}