Golang简单实现 分布式缓存+一致性哈希+节点再平衡(gossip + consistent + rebalance)

#----------------基于HTTP----------------#
# 查看状态
curl 127.0.0.1:12345/status
# 插入一个kv
curl -v  127.0.0.1:12345/cache/testkey -XPUT -d testvalue
# 查看key的val
curl 127.0.0.1:12345/cache/testkey
# 查看状态
curl 127.0.0.1:12345/status
# 删除key
curl 127.0.0.1:12345/cache/testkey -XDELETE
# 查看状态
curl 127.0.0.1:12345/status

func (s *Server) Listen() {
  //这里就是对一个map进行增删改查，并维护一个stat
  http.Handle("/cache/", s.cacheHandler())
  //返回上面维护的sata结构体
  http.Handle("/status", s.statusHandler())
  //consistent实现Members，以切片形式返回所有活跃节点的地址
  //m := h.Members()
  //bytes, err := json.Marshal(m)
  //w.Write(bytes)
  http.Handle("/cluster", s.clusterHandler())
  //最后介绍
  http.Handle("/rebalance", s.rebalanceHandler())
  http.ListenAndServe(s.Addr()+":12345", nil)
}

for {
  conn, err := listener.Accept()
  if err != nil {
    log.Println(err)
  }
  //对每一个连接都开一个process的协程
  go s.process(conn)
}
//代码中忽略了细节，可能会与源码有出路
func (s *Server) process(conn net.Conn) {
  r := bufio.NewReader(conn)
  for {
    op, err := r.ReadByte()
    switch op {
    case 'S':
      err = s.set(conn, r)
    case 'G':
      err = s.get(conn, r)
    case 'D':
      err = s.del(conn, r)
  }
}
//下面以get为例，set和del原理一样
func (s *Server) get(conn net.Conn, r *bufio.Reader) error {
  key, err := s.readKey(r)
  if err != nil {
    return sendResponse(nil, err, conn)
  }
  val, err := s.Get(key)
  return sendResponse(val, err, conn)
}
//通过readKey对TCP字节流进行解析
func (s *Server) readKey(r *bufio.Reader) (string, error) {
  keyLen, err := readLen(r)
  key := make([]byte, keyLen)
  _, err = io.ReadFull(r, key)
  //这里进行一致性哈希运算，判断该key是否映射到该节点上
  //如果是则继续运行，如果不是则返回重定向到哪个节点
  jumpAddr, ok := s.ShouldProcess(string(key))
  if !ok {
    return "", errors.New("redirect " + jumpAddr)
  }
  return string(key), nil
}
//这里进行一致性哈希运算，判断该key是否映射到该节点上
func (n *node) ShouldProcess(key string) (string, bool) {
  addr, _ := n.Get(key)
  return addr, addr == n.addr
}

//创建新节点
node, err := cluster.New(*nodeAddr, *cls)
func New(addr, cluster string) (Node, error) {
  //创建gossip新节点的config
  config := memberlist.DefaultLANConfig()
  config.Name = addr
  config.BindAddr = addr
  config.LogOutput = ioutil.Discard
  //创建新节点
  mbl, err := memberlist.Create(config)
  if err != nil {
    return nil, err
  }
  if cluster == "" {
    cluster = addr
  }
  existing := []string{cluster}
  //连接到集群
  _, err = mbl.Join(existing)
  //创建一致性哈希的节点实例
  circle := consistent.New()
  //设置虚拟节点数量
  circle.NumberOfReplicas = 256
  go func() {
    for {
      //获取集群成员
      m := mbl.Members()
      nodes := make([]string, len(m))
      for i, n := range m {
        nodes[i] = n.Name
      }
      //每隔1s将集群节点列表m更新到circle中
      circle.Set(nodes)
      time.Sleep(time.Second)
    }
  }()
  return &node{circle, addr}, nil
}

//创建一致性哈希的节点实例
circle := consistent.New()
//设置虚拟节点数量
circle.NumberOfReplicas = 256
//每隔1s将集群节点列表m更新到circle中
circle.Set(nodes)
//这里进行一致性哈希运算，判断该key是否映射到该节点上
addr, _ := n.Get(key)

func (h *rebalanceHandler) rebalance() {
  s := h.NewScanner()
  defer s.Close()
  client := &http.Client{}
  //遍历本节点所有的key
  for s.Scan() {
    k := s.Key()
    redirectAddr, ok := h.ShouldProcess(k)
    if !ok {//如果因为扩缩容，该key不再映射到本节点
      //则将key写入到对应的节点去
      r, _ := http.NewRequest(http.MethodPut, "http://"+redirectAddr+":12345/cache/"+k, bytes.NewReader(s.Value()))
      client.Do(r)
      h.Del(k)
    }
  }
}