基于gRPC的注册发现与负载均衡的原理和实战

// Register 注册自定义resolver
func Register(b Builder) {
    m[b.Scheme()] = b
}
// Builder 定义resolver builder
type Builder interface {
    Build(target Target, cc ClientConn, opts BuildOptions) (Resolver, error)
    Scheme() string
}

type Target struct {
    Scheme    string // 表示要使用的名称系统
    Authority string // 表示一些特定于方案的引导信息
    Endpoint  string // 指出一个具体的名字
}

type Resolver interface {
    ResolveNow(ResolveNowOptions)
    Close()
}

// BuildDiscovTarget 构建target
func BuildDiscovTarget(endpoints []string, key string) string {
    return fmt.Sprintf("%s://%s/%s", resolver.DiscovScheme,
        strings.Join(endpoints, resolver.EndpointSep), key)
}
// RegisterResolver 注册自定义的Resolver
func RegisterResolver() {
    resolver.Register(&dirBuilder)
    resolver.Register(&disBuilder)
}

func (d *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (
    resolver.Resolver, error) {
    hosts := strings.FieldsFunc(target.Authority, func(r rune) bool {
        return r == EndpointSepChar
    })
  // 获取服务列表
    sub, err := discov.NewSubscriber(hosts, target.Endpoint)
    if err != nil {
        return nil, err
    }
    update := func() {
        var addrs []resolver.Address
        for _, val := range subset(sub.Values(), subsetSize) {
            addrs = append(addrs, resolver.Address{
                Addr: val,
            })
        }
    // 调用UpdateState方法更新
        cc.UpdateState(resolver.State{
            Addresses: addrs,
        })
    }
  // 添加监听，当服务地址发生变化会触发更新
    sub.AddListener(update)
  // 更新服务列表
    update()
    return &nopResolver{cc: cc}, nil
}

func (b *p2cPickerBuilder) Build(readySCs map[resolver.Address]balancer.SubConn) balancer.Picker {
    if len(readySCs) == 0 {
        return base.NewErrPicker(balancer.ErrNoSubConnAvailable)
    }
    var conns []*subConn
    for addr, conn := range readySCs {
        conns = append(conns, &subConn{
            addr:    addr,
            conn:    conn,
            success: initSuccess,
        })
    }
    return &p2cPicker{
        conns: conns,
        r:     rand.New(rand.NewSource(time.Now().UnixNano())),
        stamp: syncx.NewAtomicDuration(),
    }
}

func (p *p2cPicker) Pick(ctx context.Context, info balancer.PickInfo) (
    conn balancer.SubConn, done func(balancer.DoneInfo), err error) {
    p.lock.Lock()
    defer p.lock.Unlock()
    var chosen *subConn
    switch len(p.conns) {
    case 0:
        return nil, nil, balancer.ErrNoSubConnAvailable // 没有可用链接
    case 1:
        chosen = p.choose(p.conns[0], nil) // 只有一个链接
    case 2:
        chosen = p.choose(p.conns[0], p.conns[1])
    default: // 选择一个健康的节点
        var node1, node2 *subConn
        for i := 0; i < pickTimes; i++ {
            a := p.r.Intn(len(p.conns))
            b := p.r.Intn(len(p.conns) - 1)
            if b >= a {
                b++
            }
            node1 = p.conns[a]
            node2 = p.conns[b]
            if node1.healthy() && node2.healthy() {
                break
            }
        }
        chosen = p.choose(node1, node2)
    }
    atomic.AddInt64(&chosen.inflight, 1)
    atomic.AddInt64(&chosen.requests, 1)
    return chosen.conn, p.buildDoneFunc(chosen), nil
}