解析grpc架构与原理

rpc Echo(EchoRequest) returns (EchoResponse){
}

rpc Echo(EchoRequest) returns (stream EchoResponse){
}

rpc Echo(stream EchoRequest) returns (EchoResponse){
}

rpc Echo(stream EchoRequest) returns (stream EchoResponse){
}

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error)

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
	cc := &ClientConn{
		target:            target,
		csMgr:             &connectivityStateManager{},
		conns:             make(map[*addrConn]struct{}),
		dopts:             defaultDialOptions(),
		blockingpicker:    newPickerWrapper(),
		czData:            new(channelzData),
		firstResolveEvent: grpcsync.NewEvent(),
	}
         // …

	// Determine the resolver to use.
        // 根据target字符串，解析出结构体出来(Scheme, Authority, Endpoint)， 如果不是按照标准的target格式，则当做target为Endpoint, 如host:port的格式.
	cc.parsedTarget = grpcutil.ParseTarget(cc.target, cc.dopts.copts.Dialer != nil)
	channelz.Infof(logger, cc.channelzID, "parsed scheme: %q", cc.parsedTarget.Scheme)
        // 根据scheme拿到对应的解析器创建者(resolverBuilder), 这些resolver 是在定义具体的resolver的时候注册到grpc里的。
	resolverBuilder := cc.getResolver(cc.parsedTarget.Scheme)
        // 如果没有拿到对应的resolverBuilder, 尝试用默认的，也就是passthrough的方式
        // 这里用到创建者模式，将创建resolver交给resolverBuilder, 让resolver专注业务逻辑，创建的参数交给resolverBuiler
	if resolverBuilder == nil {
		// If resolver builder is still nil, the parsed target's scheme is
		// not registered. Fallback to default resolver and set Endpoint to
		// the original target.
		channelz.Infof(logger, cc.channelzID, "scheme %q not registered, fallback to default scheme", cc.parsedTarget.Scheme)
		cc.parsedTarget = resolver.Target{
			Scheme:   resolver.GetDefaultScheme(),
			Endpoint: target,
		}
		resolverBuilder = cc.getResolver(cc.parsedTarget.Scheme)
		if resolverBuilder == nil {
			return nil, fmt.Errorf("could not get resolver for default scheme: %q", cc.parsedTarget.Scheme)
		}
	}

        // …..
        // 负载均衡创建者的一些参数
	cc.balancerBuildOpts = balancer.BuildOptions{
		DialCreds:        credsClone,
		CredsBundle:      cc.dopts.copts.CredsBundle,
		Dialer:           cc.dopts.copts.Dialer,
		CustomUserAgent:  cc.dopts.copts.UserAgent,
		ChannelzParentID: cc.channelzID,
		Target:           cc.parsedTarget,
	}

	// Build the resolver.
        // 这里创建一个resolverWraper, 里面包了一个resolver, 这里用了策略模式，将一些流程上的逻辑统一放到wraper里，这样让resolver的职责更单一。
	rWrapper, err := newCCResolverWrapper(cc, resolverBuilder)
	if err != nil {
		return nil, fmt.Errorf("failed to build resolver: %v", err)
	}
	cc.mu.Lock()
	cc.resolverWrapper = rWrapper
	cc.mu.Unlock()
        //…
	return cc, nil
}

func newCCResolverWrapper(cc *ClientConn) (*ccResolverWrapper, error) {
	rb := cc.dopts.resolverBuilder
	if rb == nil {
		return nil, fmt.Errorf("could not get resolver for scheme: %q", cc.parsedTarget.Scheme)
	}

	ccr := &ccResolverWrapper{
		cc:     cc,
		addrCh: make(chan []resolver.Address, 1),
		scCh:   make(chan string, 1),
	}

	var err error
	ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, resolver.BuildOption{DisableServiceConfig: cc.dopts.disableServiceConfig})
	if err != nil {
		return nil, err
	}
	return ccr, nil
}

type ClientConn interface {
	// UpdateState updates the state of the ClientConn appropriately.
	UpdateState(State) error
	// ReportError notifies the ClientConn that the Resolver encountered an
	// error.  The ClientConn will notify the load balancer and begin calling
	// ResolveNow on the Resolver with exponential backoff.
	ReportError(error)
	// NewAddress is called by resolver to notify ClientConn a new list
	// of resolved addresses.
	// The address list should be the complete list of resolved addresses.
	//
	// Deprecated: Use UpdateState instead.
	NewAddress(addresses []Address)
	// NewServiceConfig is called by resolver to notify ClientConn a new
	// service config. The service config should be provided as a json string.
	//
	// Deprecated: Use UpdateState instead.
	NewServiceConfig(serviceConfig string)
	// ParseServiceConfig parses the provided service config and returns an
	// object that provides the parsed config.
	ParseServiceConfig(serviceConfigJSON string) *serviceconfig.ParseResult
}

func (r *TLResolveBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOption) (resolver.Resolver, error) {
    grpclog.Infof("build tl resolver %s", target.Endpoint)

   // 这里解析拿到对应的service名
    service, err := parseTLAddress(target.Endpoint)
    if err != nil {
        return nil, fmt.Errorf("bulid tl resolver failed %v", err)
    }

	// 初始化路由表
	_, err = tl.ApiGetRoute(service)
	if err != nil {
		return nil, fmt.Errorf("get route(service:%s) failed %v", service, err)
	}

    tlResolver := TLResolver{
       service:     service,
        cc:              cc,
    }
   // 起一个协程，定期从名字服务(或者本地agent)获取最新目标服务地址列表, 并更新到链接池里。
    go tlResolver.watcher()

    return &tlResolver, nil
}

func (r *TLResolver) watcher() {
    for {
        r.updateAddress()

        // 每秒钟获取一次tl服务地址
        time.Sleep(time.Second)
    }
}

func (r *TLResolver) updateAddress() {
    // 获取最新的地址列表
    destList, err := tl.ApiGetRouteTable(r.service)
    if err != nil {
        grpclog.Errorf("get tl failed %s", err)
        return
    }

   // 这里不用担心重复地址问题, 在clientconn.go的handleResolvedAddrs中会判断, 如果地址完全一致则会略过
    if len(destList) > 0 {
        r.cc.NewAddress(genAddrStr(destList))
    }
}

func (b *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
	s := state.ConnectivityState
        // …
	b.scStates[sc] = s
	switch s {
	case connectivity.Idle:
		sc.Connect()
	case connectivity.Shutdown:
		// When an address was removed by resolver, b called RemoveSubConn but
		// kept the sc's state in scStates. Remove state for this sc here.
		delete(b.scStates, sc)
	}

	oldAggrState := b.state
	b.state = b.csEvltr.RecordTransition(oldS, s)

         // 重新生成picker
	// Regenerate picker when one of the following happens:
	//  - this sc became ready from not-ready
	//  - this sc became not-ready from ready
	//  - the aggregated state of balancer became TransientFailure from non-TransientFailure
	//  - the aggregated state of balancer became non-TransientFailure from TransientFailure
       	if (s == connectivity.Ready) != (oldS == connectivity.Ready) ||
		(b.state == connectivity.TransientFailure) != (oldAggrState == connectivity.TransientFailure) {
		b.regeneratePicker()
	}

        // 更新balancer的状态
	b.cc.UpdateBalancerState(b.state, b.picker)
}

// 重新根据新地址生成选择器
func (b *baseBalancer) regeneratePicker() {
	if b.state == connectivity.TransientFailure {
		b.picker = NewErrPicker(balancer.ErrTransientFailure)
		return
	}
	readySCs := make(map[resolver.Address]balancer.SubConn)

	// 选出所有处于ready状态的subConn
	for addr, sc := range b.subConns {
		if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
			readySCs[addr] = sc
		}
	}
	b.picker = b.pickerBuilder.Build(readySCs)
}

package TL_resolver

import (
    "google.golang.org/grpc/resolver"
    "fmt"
    "time"
    "strings"
    "strconv"
    "google.golang.org/grpc/grpclog"
)

type TLResolveBuilder struct {

}

type TLResolver struct {
    service   string
    cc          resolver.ClientConn
}

func NewBuilder() resolver.Builder {
    return &TLResolveBuilder {
    }
}

func (r *TLResolver) watcher() {
    for {
        r.updateAddress()

        // 每秒钟获取一次TL服务地址
        time.Sleep(time.Second)
    }
}

func (r *TLResolver) updateAddress() {
    destList, err := TL.ApiGetRouteTable(r.modId, r.cmdId)
    if err != nil {
        grpclog.Errorf("get TL failed %s", err)
        return
    }

    // 当刚装上TL_agent时，第一次会返回空，所以重试一次
    if destList == nil || len(destList) == 0 {
        destList, err = TL.ApiGetRouteTable(r.modId, r.cmdId)
        if err != nil {
            grpclog.Errorf("get TL failed %s", err)
            return
        }
        if destList == nil || len(destList) == 0 {
            grpclog.Infof("retry get-route-table failed! destList is empty, modid:%d, cmdid:%d", r.modId, r.cmdId)
        }
    }

    // 这里不用担心重复地址问题, 在clientconn.go的handleResolvedAddrs中会判断, 如果地址完全一致则会略过
    if len(destList) > 0 {
        r.cc.NewAddress(genAddrStr(destList))
    }
}

func genAddrStr(destList []TL.Dest) []resolver.Address {
    addrList := make([]resolver.Address, 0)
    for _, item := range destList {
        addrStr := fmt.Sprintf("%s:%d", item.Ip, item.Port)
        addrList = append(addrList, resolver.Address{Addr: addrStr})
    }

    return addrList
}


func (r *TLResolver) ResolveNow(opts resolver.ResolveNowOption) {
    r.updateAddress()
}

func (r *TLResolver) Close() {

}


func (r *TLResolveBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOption) (resolver.Resolver, error) {
    grpclog.Infof("build TL resolver %s", target.Endpoint)

    service := target.Endpoint
    if err != nil {
        return nil, fmt.Errorf("bulid TL resolver failed %v", err)
    }

    // init route table
    _, err = TL.ApiGetRoute(service)
    if err != nil {
        return nil, fmt.Errorf("get route failed %v", err)
    }

    TLResolver := TLResolver{
        service:          service,
        cc:              cc,
    }

    go TLResolver.watcher()

    return &TLResolver, nil
}


func (r *TLResolveBuilder) Scheme() string {
    return "TL"
}

package TL_balancer

import (
	"fmt"
	"golang.org/x/net/context"
	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/resolver"
	"strconv"
	"sync"
	"time"
	"google.golang.org/grpc/grpclog"
	"google.golang.org/grpc/balancer/base"
)

func NewBalancerBuilder(service string) balancer.Builder {
	return base.NewBalancerBuilder(getBalancerName(modId, cmdId), &TLPickerBuilder{
             service: service
	})
}

func getBalancerName(service string) string {
	return “TL”
}

type TLPickerBuilder struct {
     service string
}

func (b *TLPickerBuilder) Build(readySCs map[resolver.Address]balancer.SubConn) balancer.Picker {
	grpclog.Infof("TLPickerBuilder: newPicker called with readySCs: %v", readySCs)

	connMap := make(map[string]balancer.SubConn)
	connSlice := make([]balancer.SubConn, 0)
	for addr, sc := range readySCs {
		connMap[addr.Addr] = sc
		connSlice = append(connSlice, sc)
	}

	return &TLPicker{
		connMap:   connMap,
		connSlice: connSlice,
		service:     b.service,
		mu:        sync.Mutex{},
		next:      0,
	}
}

type TLPicker struct {
	connMap   map[string]balancer.SubConn
	connSlice []balancer.SubConn

        Service string
	mu   sync.Mutex
	next int
}

func (p *TLPicker) Pick(ctx context.Context, opts balancer.PickOptions) (balancer.SubConn, func(balancer.DoneInfo), error) {
	if len(p.connSlice) <= 0 {
		return nil, nil, balancer.ErrNoSubConnAvailable
	}

	// 先通过TL选址, 这样就可以复用TL的负载均衡权重
	res, err := TL.ApiGetRoute(p.service)

	if err == nil {
		if conn, ok := p.connMap[res.Ip()+":"+strconv.Itoa(res.Port())]; ok {

			// TL 调用结果上报TL
			beginTime := time.Now().UnixNano()
			doneFunc := func(dinfo balancer.DoneInfo) {
				timeSpan := (time.Now().UnixNano() - beginTime) / 1e6
				if dinfo.Err == nil {
					TL.ApiRouteResultUpdate(res,1, uint64(timeSpan))
				} else {
					TL.ApiRouteResultUpdate(res,-1, uint64(timeSpan))
				}
			}

			return conn, doneFunc, nil
		} else {
			grpclog.Warningf("Can not find addr in pool %s %d", res.Ip(), res.Port())
		}
	} else {
		grpclog.Errorf("get TL failed %s", err)
	}

	// 当通过TL选址的地址 在连接池中不存在, 则round robin返回一个连接池中的连接
	p.mu.Lock()
	sc := p.connSlice[p.next]
	p.next = (p.next + 1) % len(p.connSlice)
	p.mu.Unlock()

	return sc, nil, nil
}

resolver.Register(tl_resolver.NewBuilder())

var service = “grpc.testservice”

balancerBuilder := tl_balancer.NewBalancerBuilder(service)
balancer.Register(balancerBuilder)