怎么实现一个分布式kv系统(2)-静态分区

摘要

本节要实现的有2点

解析toml文件
计算key的hash值
将请求路由到对应的shard

编程实现

1. 定义&解析toml规则文件

定义sharding.toml文件

[[shards]]
name = "Moscow"
id = 0

[[shards]]
name = "Minsk"
id = 1

[[shards]]
name = "Kiev"
id = 2

导入解析toml的包

添加configFile参数解析

在main中读取配置文件

import (
  "github.com/BurntSushi/toml"
)

var (
  configFile = flag.String("config-file", "sharding.toml", "The configuration file")
)

func main() {
  flag.Parse()

  var c config.Config
  if _, err := toml.DecodeFile(*configFile, &c); err != nil {
    log.Fatalf("failed to decode config file(%q):%v", *configFile, err)
  }
}

2. 添加config模块

创建config/config.go

定义Config & Shard结构

package config

type Shard struct {
  Name string
  Id   string
}

type Config struct {
  Shards []Shard
}

3. 指定shard

先不自动shard，先手动指定shard

添加shard参数解析

检查shard是否存在

确认shard的id

// main.go
var (
  shardName   = flag.String("shardName", "Moscow", "The name of the shard")
)

ok, shard := c.ExistsShard(*shardName)
if !ok {
  log.Fatalf("shard %v not exists", shardName)
}

// config.go
func (c *Config) ExistsShard(name string) (bool, Shard) {
  var rc Shard
  for _, shard := range c.Shards {
    if shard.Name == name {
      return true, shard
    }
  }
  return false, rc
}

4. 写入数据到指定shard

写入规则：hash(key) % shardCounter 就是需要写入的分片

更新Server结构，添加shardCounter&shardIndex字段

// web.go
type Server struct {
  db           *db.Database
  shardCounter int
  shardIdex    int
}

func NewServer(db *db.Database, shardCounter, shardIdex int) *Server {
  return &Server{
    db:           db,
    shardCounter: shardCounter,
    shardIndex:    shardIndex,
  }
}

导入计算hash的包: hash/fnv包，计算shard

更新GetHandler、SetHandler，调用getShard来获取应该写入数据的分片。

// web.go
func (s *Server) getShard(key string) uint64 {
  h := fnv.New64()
  h.Write([]byte(key))
  return h.Sum64() % uint64(s.shardCounter)
}

func (s *Server) GetHandler(w http.ResponseWriter, r *http.Request) {
  r.ParseForm()
  key := r.Form.Get("key")
  value, err := s.db.GetKey(key)
  fmt.Fprintf(w, "%q:%q; shard: %d; %v Get Called\n", key, value, s.getShard(key), err)
}

func (s *Server) SetHandler(w http.ResponseWriter, r *http.Request) {
  r.ParseForm()
  key := r.Form.Get("key")
  value := r.Form.Get("value")
  err := s.db.SetKey(key, []byte(value))
  fmt.Fprintf(w, "%q:%q; shard: %d; %v; Set called\n", key, value, s.getShard(key), err)
}

4.获取所有节点的地址

以上虽然可以计算出分片，可以还没有办法路由给其他的分片。

首先，需要知道其他分片的地址。

其次，将信息传入到Server对象中。

接着，定义redirect函数，请求转发。

最后，改造GetHandle、GetHandle转发请求。

// config.go
func (c *Config) GetAddress() map[int]string {
  addrs := make(map[int]string)
  for _, addr := range c.Shards {
    addrs[addr.Id] = addr.Address
  }
  return addrs
}

// web.go
type Server struct {
  db           *db.Database
  shardCounter int
  shardIndex   int
  addrs        map[int]string
}

// main.go
addrs := c.GetAddress()
svr := web.NewServer(db, len(c.Shards), shard.Id, addrs)

// web.go
func (s *Server) redirect(w http.ResponseWriter, r *http.Request, shard int) error {
  resp, err := http.Get("http://" + s.addrs[shard] + r.RequestURI)
  if err != nil {
    w.WriteHeader(http.StatusInternalServerError)
    fmt.Fprintf(w, "Error redirect request:%v", err)
    return err
  }
  defer resp.Body.Close()
  io.Copy(w, resp.Body)
  return nil
}



func (s *Server) GetHandler(w http.ResponseWriter, r *http.Request) {
  r.ParseForm()
  key := r.Form.Get("key")
  shard := s.getShard(key)
  if shard != uint32(s.shardIndex) {
    err := s.redirect(w, r, int(shard))
    if err != nil {
      return
    } 
  } else {
    value, err := s.db.GetKey(key)
    fmt.Fprintf(w, "%q:%q; target shard:%d; current shard:%d; %v Get Called\n", key, value, shard, s.shardIndex, err)
  }
}

func (s *Server) SetHandler(w http.ResponseWriter, r *http.Request) {
  r.ParseForm()
  key := r.Form.Get("key")
  value := r.Form.Get("value")
  shard := s.getShard(key)
  if shard != uint32(s.shardIndex) {
    err := s.redirect(w, r, int(shard))
    if err != nil {
      return
    }
  } else {
    err := s.db.SetKey(key, []byte(value))
    fmt.Fprintf(w, "%q:%q; target shard:%d; current shard:%d; %v Set Called\n", key, value, shard, s.shardIndex, err)
  }
}

5.测试

get(){
    for key in a b c;do 
        echo -e "\n========= get $key========="
        for port in 8000 8001 8002;do 
            curl "http://127.0.0.1:$port/get?key=$key"
        done
    done
}

set(){
    for key in a b c;do 
        echo -e "\n========= set $key========="
        for port in 8000 8001 8002;do 
            curl "http://127.0.0.1:$port/set?key=$key&value=$key"
        done
    done
}

set
get