groupcache源码(1) HTTPPool

groupcache调用流程

• 第一步就是创建了HTTPPool这个对象

peers := groupcache.NewHTTPPool("http://" + local_addr)
peers.Set(peers_addrs...)
cache := groupcache.NewGroup()
http.HandleFunc()

HTTPOOL对象

type HTTPPool struct {
    Context func(*http.Request) Context
    Transport func(Context) http.RoundTripper

    self string
    opts HTTPPoolOptions

    peers       *consistenthash.Map
    httpGetters map[string]*httpGetter // keyed by e.g. "http://10.0.0.2:8008"
}

func (p *HTTPPool) Set()
func (p *HTTPPool) PickPeer(()
func (p *HTTPPool) ServeHTTP()

NewHTTPPool()

func NewHTTPPool(self string) *HTTPPool {
    // 初始化HTTPPool
    p := NewHTTPPoolOpts(self, nil)
    // 注册路由
    http.Handle(p.opts.BasePath, p)
    return p
}

NewHTTPPoolOpts()

// NewHTTPPoolOpts() 使用self参数初始化一个 HTTPPool对象
func NewHTTPPoolOpts(self string, o *HTTPPoolOptions) *HTTPPool {
    if httpPoolMade {
        panic("groupcache: NewHTTPPool must be called only once")
    }
    // httpPoolMade 确保HTTPPool唯一的标识
    httpPoolMade = true

    p := &HTTPPool{
        self:        self,
        httpGetters: make(map[string]*httpGetter),
    }
    if o != nil {
        p.opts = *o
    }
    if p.opts.BasePath == "" {
        p.opts.BasePath = defaultBasePath
    }
    if p.opts.Replicas == 0 {
        p.opts.Replicas = defaultReplicas
    }

    // 设置副本数量和HachFunc 创建一致性哈希对象
    p.peers = consistenthash.New(p.opts.Replicas, p.opts.HashFn)

    // 注册peers.portPicker
    RegisterPeerPicker(func() PeerPicker { return p })
    return p
}

• 首先是用配置参数初始化
• consistenthash是groupcache包中负责提供一致性哈希的模块

consistenthash 一致性Hashap

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

func New(replicas int, fn Hash) *Map {
    m := &Map{
        replicas: replicas,
        hash:     fn,
        hashMap:  make(map[int]string),
    }
    if m.hash == nil {
        m.hash = crc32.ChecksumIEEE
    }
    return m
}

func (m *Map) IsEmpty() bool
func (m *Map) Add(keys ...string)
func (m *Map) Get(key string) string 

• 通过 consistenthash.New()的调用获取一个一致性hashMap
• 此HashMap可使用IsEmpty，Add，Get等方法

RegisterPeerPicker 注册匹配规则

// PickPeer()方法在HTTPPool中有实现
func (p *HTTPPool) PickPeer(key string) (ProtoGetter, bool) {
    p.mu.Lock()
    defer p.mu.Unlock()
    if p.peers.IsEmpty() {
        return nil, false
    }
    if peer := p.peers.Get(key); peer != p.self {
        return p.httpGetters[peer], true
    }
    return nil, false
}

// PeerPicker参数是接口类型，传入的是HTTPPool.PickPeer()
// 用传入的函数来初始化portPicker这个全局变量
func RegisterPeerPicker(fn func() PeerPicker) {
    if portPicker != nil {
        panic("RegisterPeerPicker called more than once")
    }
    portPicker = func(_ string) PeerPicker { return fn() }
}

• 使用HTTPPool对象p.PickPeer()来初始化portPicker
• PeerPicker参数是接口类型，PickPeer()方法在HTTPPool中有实现

http.Handle()

• http.Handle(pattern string, handler Handler)
• http是net/http包，这里Handle()是在设置路由，HTTPPOOL实现了ServeHTTP()这个接口函数

func (p *HTTPPool) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // 处理url获取groupName，keyname
    groupName := parts[0]
    key := parts[1]

    // Fetch the value for this group/key.
    group := GetGroup(groupName)
    if group == nil {
        http.Error(w, "no such group: "+groupName, http.StatusNotFound)
        return
    }
    var ctx Context
    if p.Context != nil {
        ctx = p.Context(r)
    }

    group.Stats.ServerRequests.Add(1)
    var value []byte
    err := group.Get(ctx, key, AllocatingByteSliceSink(&value))
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

    // Write the value to the response body as a proto message.
    body, err := proto.Marshal(&pb.GetResponse{Value: value})
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    w.Header().Set("Content-Type", "application/x-protobuf")
    w.Write(body)
}

• TODO：groupMap由 NewGroup()初始化，路由注册原理
• 至此
  • 初始化HTTPPool这个唯一对象
  • 生成了一致性HashMap处理对象，带IsEmpty，Add，Get等方法
  • 使用net/http包注册路由规则