实现使用rpc进行节点间获取缓存；并使用etcd实现缓存服务的注册与发现；节点间通信使用protobuf格式

.gitignore

@@ -7,3 +7,6 @@
 # Macos file system
 .DS_Store
 */.DS_Store
+
+# etcd data
+default.etcd

cache.go

@@ -7,9 +7,8 @@ package peanutcache
 // cache 模块负责提供对lru模块的并发控制
 
 import (
- "sync"
-
  "github.com/peanutzhen/peanutcache/lru"
+ "sync"
 )
 
 // 这样设计可以进行cache和算法的分离，比如我现在实现了lfu缓存模块

client.go

@@ -5,47 +5,52 @@
 package peanutcache
 
 import (
+ "context"
  "fmt"
- "io/ioutil"
- "net/http"
- "net/url"
+ pb "github.com/peanutzhen/peanutcache/peanutcachepb"
+ "github.com/peanutzhen/peanutcache/registry"
+ "time"
+
+ clientv3 "go.etcd.io/etcd/client/v3"
 )
 
-// client 模块实现peanutcache访问其他远程节点
-// 从而获取缓存的能力
+// client 模块实现peanutcache访问其他远程节点 从而获取缓存的能力
 
-type Client struct {
- reqURL string
+type client struct {
+ name string // 服务名称 pcache/ip:addr
 }
 
 // Fetch 从remote peer获取对应缓存值
-func (c *Client) Fetch(group string, key string) ([]byte, error) {
- // 构造请求url
- u := fmt.Sprintf(
- "%s%s/%s",
- c.reqURL,
- url.QueryEscape(group),
- url.QueryEscape(key),
- )
-
- resp, err := http.Get(u)
+func (c *client) Fetch(group string, key string) ([]byte, error) {
+ // 创建一个etcd client
+ cli, err := clientv3.New(defaultEtcdConfig)
  if err != nil {
  return nil, err
  }
- defer resp.Body.Close()
-
- if resp.StatusCode != http.StatusOK {
- return nil, fmt.Errorf("peer Statuscode: %d", resp.StatusCode)
+ defer cli.Close()
+ // 发现服务 取得与服务的连接
+ conn, err := registry.EtcdDial(cli, c.name)
+ if err != nil {
+ return nil, err
  }
- body, err := ioutil.ReadAll(resp.Body)
+ defer conn.Close()
+ grpcClient := pb.NewPeanutCacheClient(conn)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ resp, err := grpcClient.Get(ctx, &pb.GetRequest{
+ Group: group,
+ Key: key,
+ })
  if err != nil {
- return nil, fmt.Errorf("read response body failed, %v", err)
+ return nil, fmt.Errorf("could not get %s/%s from peer %s", group, key, c.name)
  }
- return body, nil
+
+ return resp.GetValue(), nil
 }
 
-func NewClient(reqURL string) *Client {
- return &Client{reqURL: reqURL}
+func NewClient(service string) *client {
+ return &client{name: service}
 }
 
-var _ Fetcher = (*Client)(nil)
+// 测试Client是否实现了Fetcher接口
+var _ Fetcher = (*client)(nil)

go.mod

@@ -1,3 +1,9 @@
 module github.com/peanutzhen/peanutcache
 
 go 1.16
+
+require (
+ go.etcd.io/etcd/client/v3 v3.5.0
+ google.golang.org/grpc v1.38.0
+ google.golang.org/protobuf v1.27.1
+)

peanutcache.go

@@ -74,6 +74,16 @@ func GetGroup(name string) *Group {
  return g
 }
 
+func DestroyGroup(name string) {
+ g := GetGroup(name)
+ if g != nil {
+ svr := g.server.(*server)
+ svr.Stop()
+ delete(groups, name)
+ log.Printf("Destroy cache [%s %s]", name, svr.addr)
+ }
+}
+
 func (g *Group) Get(key string) (ByteView, error) {
  if key == "" {
  return ByteView{}, fmt.Errorf("key required")
@@ -90,7 +100,7 @@ func (g *Group) load(key string) (ByteView, error) {
  view, err := g.flight.Fly(key, func() (interface{}, error) {
  if g.server != nil {
  if fetcher, ok := g.server.Pick(key); ok {
- bytes, err := fetcher.Fetch(g.name, key);
+ bytes, err := fetcher.Fetch(g.name, key)
  if err == nil {
  return ByteView{b: cloneBytes(bytes)}, nil
  }

peanutcachepb/peanutcache.proto

@@ -0,0 +1,19 @@
+syntax = "proto3";
+
+package peanutcachepb;
+
+option go_package = "github.com/peanutcache/peanutcachepb";
+
+message GetRequest {
+ string group = 1;
+ string key = 2;
+}
+
+message GetResponse {
+ bytes value = 1;
+}
+
+service PeanutCache {
+ rpc Get(GetRequest) returns (GetResponse);
+}
+