grgoupcache源码走读（三）:groupcache的特性和使用案例

    了解了前面两篇基于本地缓存的策略以后，咱们再来看看分布式缓存groupcache。它的做者也是memcache的做者，在github中对groupcache的简介有以下一句：

groupcache is a caching and cache-filling library, intended as a replacement for memcached in many cases.

    直译过来就是groupcache是一个kv缓存库，而且致力于在某些场景下替代mc。但愿读者经过接下来的内容能对groupcache的使用场景有一些本身的想法。    

    笔者简单总结了一下它应该具有的一些特性，若是有不全的欢迎你们补充：

• 只是一个代码包，能够直接使用，不须要单独配置服务器，既是客户端也是服务器端，这主要是由于它既可能直接返回请求的key的结果，也可能将请求转给其余peer来处理。对访问的用户来讲，它是服务端，对其余peer来讲它又是客户端（稍微有点绕，不明白的看以后的demo吧）；
• 分布式缓存，支持一致性哈希，这里主要是经过peers和consistenthash实现的；
• 采用的缓存机制是LRU，LRU主要经过list.List队列来实现的，不支持过时机制；（因此。。是的，若是有长期不用的数据，只能寄但愿于缓存满了以后自动剔除队尾了）
• 对于客户端或者使用用户来讲，对数据的操做只能get，不支持set，update以及delete；
• 实现了缓存过滤机制(这个概念也是笔者网上看到的)， 代码中叫singlefilght；通常状况下，缓存存在被击穿的风险，即大量相同的请求并发访问，若是此时cache miss，那么这些请求都会落地到服务器端的db上，致使db压力过大甚至最终形成整个服务不可用的状况。而singleflight所作的就是针对并发访问的相同miss key，只有一个请求会真正load data（如访问db），其他请求都会等待，最终将结果返回给其余请求，使得全部请求能返回一致的结果。
• 对热点数据进行备份，虽然是一致性哈希，可是对于过热数据，也会进行多节点备份，缓解过热数据访问对节点形成的压力；

    以上就是笔者暂时能想到的关于groupcache的特性。其中第四点，groupcache只支持get操做多是和mc差异最大的地方了，由于真正只能get的场景着实很少。。。

    使用案例

        PS:   在启动groupcache的时候须要监听两个端口，一个端口是用来外部访问groupcache的，另一个端口是集群内部peer互相通讯使用的。                     

    接下来看下如何使用groupcache：

run_peer1.go

const defaultHost = "127.0.0.1:9001"
const group_addr = ":8081"

func main() {
        if len(os.Args) <= 1 { 
                fmt.Fprintf(os.Stderr, "Usage: %s peer1 [peer2...]", os.Args[0])
                os.Exit(1)
        }   

        //本地peer地址
        self := flag.String("self", defaultHost, "self node")
        flag.Parse()

        //cache集群全部节点
        cluster := os.Args[1:]

        //初始化本地groupcache, 并监听groupcache相应的端口
        setUpGroup("test_cache")
        //本地peer
        peers := groupcache.NewHTTPPool(addrsToUrl(*self)[0])
        peers.Set(addrsToUrl(cluster...)...) //设置集群信息 用以本机缓存没命中的时候，一致性哈希查找key的存储节点, 并经过http请求访问

        selfPort := strings.Split(*self, ":")[1]
        http.ListenAndServe(":"+selfPort, peers) //监听本机集群内部通讯的端口

}

//启动groupcache
func setUpGroup(name string) {
        //缓存池,
        stringGroup := groupcache.NewGroup(name, 1<<20, groupcache.GetterFunc(func(_ groupcache.Context, key string, dest groupcache.Sink) error {
                //当cache miss以后，用来执行的load data方法
                fp, err := os.Open("groupcache.conf")
                if err != nil {
                        return err
                }
                defer fp.Close()

                fmt.Printf("look up for %s from config_file\n", key)
                //按行读取配置文件
                buf := bufio.NewReader(fp)
                for {
                        line, err := buf.ReadString('\n')
                        if err != nil {
                                if err == io.EOF {
                                        dest.SetBytes([]byte{})
                                        return nil
                                } else {
                                        return err
                                }
                        }

                        line = strings.TrimSpace(line)
                        parts := strings.Split(line, "=")
                        if len(parts) > 2 {
                                continue
                        } else if parts[0] == key {
                                dest.SetBytes([]byte(parts[1]))
                                return nil
                        } else {
                                continue
                        }
                }
        }))

        http.HandleFunc("/config", func(rw http.ResponseWriter, r *http.Request) {
                k := r.URL.Query().Get("key")
                var dest []byte
                fmt.Printf("look up for %s from groupcache\n", k)
                if err := stringGroup.Get(nil, k, groupcache.AllocatingByteSliceSink(&dest)); err != nil {
                        rw.WriteHeader(http.StatusNotFound)
                        rw.Write([]byte("this key doesn't exists"))
                } else {
                        rw.Write([]byte(dest))
                }

        })

        //可以直接访问cache的端口, 启动http服务
        //http://ip:group_addr/config?key=xxx
        go http.ListenAndServe(group_addr, nil)

}

//将ip:port转换成url的格式
func addrsToUrl(node_list ...string) []string {
        urls := make([]string, len(node_list))
        for k, addr := range node_list {
                urls[k] = "http://" + addr
        }   

        return urls
}

    这里执行 go run run_peer1.go 127.0.0.1:9001 就能启动一个groupcache实例，其中8080是对外访问端口，9001是集群内部通讯端口， 固然在这个例子中，这是一个只有单节点的集群。简单测试一下：

    执行 curl -i http://127.0.0.1:8081/config?key=name 命令两次， 能够看到第一次走到了GetterFunc类型的方法中去从配置文件load数据了，第二次直接从缓存中就拿到了数据。

        

     若是想要单机模拟多节点集群，只须要将上述代码配置稍微修改一下保存成run_peer2.go便可，修改以下：

//run_peer2.go
const defaultHost = "127.0.0.1:9002"
const group_addr = ":8082"

执行命令， 便可实现集群模式：

go run run_peer1.go 127.0.0.1:9001 127.0.0.1:9002
go run run_peer2.go 127.0.0.1:9001 127.0.0.1:9002

    此时能够观察一下整个集群的执行流程以及一致性hash，执行 curl -i http://127.0.0.1:8082/config?key=name，看一下输出结果：

        

        

    由于访问的8082端口，会先去访问8082端口，而后查找name这个键所属的存储节点，如图一；发现存储节点并不是本节点以后，经过http请求访问对应存储节点，获取数据，如图二。据此，咱们能够总结一下groupcache总的流程：

1. 输入url地址，访问url对应的存储节点front server;
2. 先查看groupcache中是否有该key，有直接返回；若无，进入第三步数据load；
3. load data步骤，实现了前文提的缓存过滤机制；这里会从新查询groupcache，防止重复load；若是尚未，会先经过一致性哈希判断key所在的存储节点cache peer；若是非本机，则进入第四步；不然，进入第五步；
4. 调用http请求，访问对应的存储节点cache peer，经过上文的集群内通讯端口(9001/9002)访问，对应的server将依次执行2,3,5步骤，返回数据。
5. 该key属于本机，则调用初始化group时定义的GetterFunc类型方法load数据，此处方法彻底自定义，既能够访问配置文件，也能够访问db。

具体的结构图以下所示，这个只画出了一半的流程，但愿你们能看明白。

            

至此，关于groupcache的总体运做流程描述完毕。可能有一些地方没有细讲，好比singleflight， 怎么实现的一致性哈希等等。。。若是有疑问欢迎拍砖留言，我和你们一块儿研究一块儿学习。。