Dubbo-go 源碼筆記（二）用戶端調用過程

作者 | 李志信

導讀：有了上一篇文章 《Dubbo-go 源碼筆記（一）Server 端開啟服務過程》 的鋪墊，可以類比用戶端啟動于服務端的啟動過程。其中最大的差別是服務端通過 zk 注冊服務，釋出自己的ivkURL并訂閱事件開啟監聽；而客戶應該是通過zk注冊元件，拿到需要調用的serviceURL，更新invoker并重寫使用者的RPCService，進而實作對遠端過程調用細節的封裝。

配置檔案和用戶端源代碼

1. client 配置檔案

helloworld 提供的 demo：profiles/client.yaml。

registries :
  "demoZk":
    protocol: "zookeeper"
    timeout  : "3s"
    address: "127.0.0.1:2181"
    username: ""
    password: ""
references:
  "UserProvider":
    # 可以指定多個registry，使用逗号隔開;不指定預設向所有注冊中心注冊
    registry: "demoZk"
    protocol : "dubbo"
    interface : "com.ikurento.user.UserProvider"
    cluster: "failover"
    methods :
    - name: "GetUser"
      retries: 3           

可看到配置檔案與之前讨論過的 Server 端非常類似，其 refrences 部分字段就是對目前服務要主調的服務的配置，其中詳細說明了調用協定、注冊協定、接口 id、調用方法、叢集政策等，這些配置都會在之後與注冊元件互動、重寫 ivk、調用的過程中使用到。

2. 用戶端使用架構源碼

user.go：

func init() {
  config.SetConsumerService(userProvider)
  hessian.RegisterPOJO(&User{})
}           

main.go：

func main() {
  hessian.RegisterPOJO(&User{})
  config.Load()
  time.Sleep(3e9)
  println("\n\n\nstart to test dubbo")
  user := &User{}
  err := userProvider.GetUser(context.TODO(), []interface{}{"A001"}, user)
  if err != nil {
      panic(err)
  }
  println("response result: %v\n", user)
  initSignal()
}           

在官網提供的 helloworld demo 的源碼中，可看到與服務端類似，在 user.go 内注冊了 rpc-service，以及需要 rpc 傳輸的結構體 user。

在 main 函數中，同樣調用了 config.Load() 函數，之後就可以通過實作好的 rpc-service：userProvider 直接調用對應的功能函數，即可實作 rpc 調用。

可以猜到，從 hessian 注冊結構、SetConsumerService，到調用函數 .GetUser() 期間，使用者定義的 rpc-service 也就是 userProvider 對應的函數被重寫，重寫後的 GetUser 函數已經包含實作了遠端調用邏輯的 invoker。

接下來，就要通過閱讀源碼，看看 dubbo-go 是如何做到的。

實作遠端過程調用

1. 加載配置檔案

// file: config/config_loader.go :Load()

// Load Dubbo Init
func Load() {
  // init router
  initRouter()
  // init the global event dispatcher
  extension.SetAndInitGlobalDispatcher(GetBaseConfig().EventDispatcherType)
  // start the metadata report if config set
  if err := startMetadataReport(GetApplicationConfig().MetadataType, GetBaseConfig().MetadataReportConfig); err != nil {
      logger.Errorf("Provider starts metadata report error, and the error is {%#v}", err)
  return
  }
  // reference config
  loadConsumerConfig()           

在 main 函數中調用了 config.Load() 函數，進而調用了 loadConsumerConfig，類似于之前講到的 server 端配置讀入函數。

在 loadConsumerConfig 函數中，進行了三步操作：

// config/config_loader.go
func loadConsumerConfig() {
    // 1 init other consumer config
    conConfigType := consumerConfig.ConfigType
    for key, value := range extension.GetDefaultConfigReader() {}
    checkApplicationName(consumerConfig.ApplicationConfig)
    configCenterRefreshConsumer()
    checkRegistries(consumerConfig.Registries, consumerConfig.Registry)
    
    // 2 refer-implement-reference
    for key, ref := range consumerConfig.References {
        if ref.Generic {
            genericService := NewGenericService(key)
            SetConsumerService(genericService)
        }
        rpcService := GetConsumerService(key)
        ref.id = key
        ref.Refer(rpcService)
        ref.Implement(rpcService)
    }
    // 3 wait for invoker is available, if wait over default 3s, then panic
    for {}
}           

1. 檢查配置檔案并将配置寫入記憶體
2. 在 for 循環内部，依次引用（refer）并且執行個體化（implement）每個被調 reference
3. 等待三秒鐘所有 invoker 就緒

其中重要的就是 for 循環裡面的引用和執行個體化，兩步操作，會在接下來展開讨論。

至此，配置已經被寫入了架構。

2. 擷取遠端 Service URL，實作可供調用的 invoker

上述的 ref.Refer 完成的就是這部分的操作。

圖（一）

1）構造注冊 url

和 server 端類似，存在注冊 url 和服務 url，dubbo 習慣将服務 url 作為注冊 url 的 sub。

// file: config/reference_config.go: Refer()
func (c *ReferenceConfig) Refer(_ interface{}) {
  //（一）配置url參數(serviceUrl)，将會作為sub
  cfgURL := common.NewURLWithOptions(
  common.WithPath(c.id),
  common.WithProtocol(c.Protocol),
  common.WithParams(c.getUrlMap()),
  common.WithParamsValue(constant.BEAN_NAME_KEY, c.id),
  )
  ...
  // （二）注冊位址可以通過url格式給定，也可以通過配置格式給定
  // 這一步的意義就是配置->提取資訊生成URL
  if c.Url != "" {// 使用者給定url資訊，可以是點對點的位址，也可以是注冊中心的位址
  // 1. user specified URL, could be peer-to-peer address, or register center's address.
  urlStrings := gxstrings.RegSplit(c.Url, "\\s*[;]+\\s*")
  for _, urlStr := range urlStrings {
    serviceUrl, err := common.NewURL(urlStr)
    ...
  }
  } else {// 配置讀入注冊中心的資訊
  //  assemble SubURL from register center's configuration mode
  // 這是注冊url，protocol = registry,包含了zk的使用者名、密碼、ip等等
  c.urls = loadRegistries(c.Registry, consumerConfig.Registries, common.CONSUMER)
  ...
  // set url to regUrls
  for _, regUrl := range c.urls {
    regUrl.SubURL = cfgURL// regUrl的subURl存目前配置url
  }
  }
  //至此，無論通過什麼形式，已經拿到了全部的regURL
  // （三）擷取registryProtocol執行個體，調用其Refer方法，傳入新建構好的regURL
  if len(c.urls) == 1 {
  // 這一步通路到registry/protocol/protocol.go registryProtocol.Refer
  // 這裡是registry
  c.invoker = extension.GetProtocol(c.urls[0].Protocol).Refer(*c.urls[0])
  } else {
  // 如果有多個注冊中心，即有多個invoker,則采取叢集政策
  invokers := make([]protocol.Invoker, 0, len(c.urls))
  ...
  }           

這個函數中，已經處理完從 Register 配置到 RegisterURL 的轉換,即圖（一）中部分：

接下來，已經拿到的 url 将被傳遞給 RegistryProtocol，進一步 refer。

2）registryProtocol 擷取到 zkRegistry 執行個體，進一步 Refer

// file: registry/protocol/protocol.go: Refer

// Refer provider service from registry center
// 拿到的是配置檔案registries的url，他能夠生成一個invoker = 指向目的addr，以供用戶端直接調用。
func (proto *registryProtocol) Refer(url common.URL) protocol.Invoker {
  var registryUrl = url
  // 這裡拿到的是referenceConfig，serviceUrl裡面包含了Reference的所有資訊，包含interfaceName、method等等
  var serviceUrl = registryUrl.SubURL
  if registryUrl.Protocol == constant.REGISTRY_PROTOCOL {// registryUrl.Proto = "registry"
  protocol := registryUrl.GetParam(constant.REGISTRY_KEY, "")
  registryUrl.Protocol = protocol//替換成了具體的值，比如"zookeeper"
  }
  // 接口對象
  var reg registry.Registry
  // （一）執行個體化接口對象，緩存政策
  if regI, loaded := proto.registries.Load(registryUrl.Key()); !loaded {
  // 緩存中不存在目前registry，建立一個reg
  reg = getRegistry(&registryUrl)
  // 緩存起來
  proto.registries.Store(registryUrl.Key(), reg)
  } else {
  reg = regI.(registry.Registry)
  }
  // 到這裡，擷取到了reg執行個體 zookeeper的registry
  //（二）根據Register的執行個體zkRegistry和傳入的regURL建立一個directory
  // 這一步存在複雜的異步邏輯，從注冊中心拿到了目的service的真實addr，擷取了invoker并放入directory，
  // 這一步将在下面詳細給出步驟
  // new registry directory for store service url from registry
  directory, err := extension.GetDefaultRegistryDirectory(&registryUrl, reg)
  if err != nil {
  logger.Errorf("consumer service %v  create registry directory  error, error message is %s, and will return nil invoker!",
    serviceUrl.String(), err.Error())
  return nil
  }
  // （三）DoRegister 在zk上注冊目前client service
  err = reg.Register(*serviceUrl)
  if err != nil {
  logger.Errorf("consumer service %v register registry %v error, error message is %s",
    serviceUrl.String(), registryUrl.String(), err.Error())
  }
  // （四）new cluster invoker，将directory寫入叢集，獲得具有叢集政策的invoker
  cluster := extension.GetCluster(serviceUrl.GetParam(constant.CLUSTER_KEY, constant.DEFAULT_CLUSTER))
  invoker := cluster.Join(directory)
  // invoker儲存
  proto.invokers = append(proto.invokers, invoker)
  return invoker
}           

可詳細閱讀上述注釋，這個函數完成了從 url 到 invoker 的全部過程：

（一）首先獲得 Registry 對象，預設是之前執行個體化的 zkRegistry，和之前 server 擷取 Registry 的處理很類似。

（二）通過構造一個新的 directory，異步拿到之前在 zk 上注冊的 server 端資訊，生成 invoker。

（三）在 zk 上注冊目前 service。

（四）叢集政策，獲得最終 invoker。

這一步完成了圖（一）中所有餘下的絕大多數操作，接下來就需要詳細地檢視 directory 的構造過程。

3）構造 directory（包含較複雜的異步操作）

圖（二）

上述的

extension.GetDefaultRegistryDirectory(&registryUrl, reg)

函數，本質上調用了已經注冊好的

NewRegistryDirectory

函數:

// file: registry/directory/directory.go: NewRegistryDirectory()

// NewRegistryDirectory will create a new RegistryDirectory
// 這個函數作為default注冊在extension上面
// url為注冊url，reg為zookeeper registry
func NewRegistryDirectory(url *common.URL, registry registry.Registry) (cluster.Directory, error) {
  if url.SubURL == nil {
  return nil, perrors.Errorf("url is invalid, suburl can not be nil")
  }
  dir := &RegistryDirectory{
  BaseDirectory:    directory.NewBaseDirectory(url),
  cacheInvokers:    []protocol.Invoker{},
  cacheInvokersMap: &sync.Map{},
  serviceType:      url.SubURL.Service(),
  registry:         registry,
  }
  dir.consumerConfigurationListener = newConsumerConfigurationListener(dir)
  go dir.subscribe(url.SubURL)
  return dir, nil
}           

首先構造了一個注冊 directory，開啟協程調用其 subscribe 函數，傳入 serviceURL。

這個 directory 目前包含了對應的 zkRegistry，以及傳入的 URL，它的 cacheInvokers 部分是空的。

進入 dir.subscribe(url.SubURL) 這個異步函數：

/ file: registry/directory/directory.go: subscribe()

// subscribe from registry
func (dir *RegistryDirectory) subscribe(url *common.URL) {
  // 增加兩個監聽，
  dir.consumerConfigurationListener.addNotifyListener(dir)
  dir.referenceConfigurationListener = newReferenceConfigurationListener(dir, url)
  // subscribe調用
  dir.registry.Subscribe(url, dir)
}           

重點來了，它調用了 zkRegistry 的 Subscribe 方法,與此同時将自己作為 ConfigListener 傳入。

我認為這種傳入 listener 的設計模式非常值得學習，而且很有 java 的味道。

針對等待 zk 傳回訂閱資訊這樣的異步操作，需要傳入一個 Listener，這個 Listener 需要實作 Notify 方法，進而在作為參數傳入内部之後，可以被異步地調用 Notify，将内部觸發的異步事件“傳遞出來”，再進一步處理加工。

層層的 Listener 事件鍊，能将傳入的原始 serviceURL 通過 zkConn 發送給 zk 服務，擷取到服務端注冊好的 url 對應的二進制資訊。

而 Notify 回調鍊，則将這串 byte[] 一步一步解析、加工；以事件的形式向外傳遞，最終落到 directory 上的時候，已經是成型的 newInvokers 了。

具體細節不再以源碼形式展示，可參照上圖查閱源碼。

至此已經拿到了 server 端注冊好的真實 invoker。

完成了圖（一）中的部分：

4）構造帶有叢集政策的 clusterinvoker

經過上述操作，已經拿到了 server 端 Invokers，放入了 directory 的 cacheinvokers 數組裡面緩存。

後續的操作對應本文從 url 到 invoker 的過程的最後一步，由 directory 生成帶有特性叢集政策的 invoker。

// （四）new cluster invoker，将directory寫入叢集，獲得具有叢集政策的invoker
  cluster := extension.GetCluster(serviceUrl.GetParam(constant.CLUSTER_KEY, constant.DEFAULT_CLUSTER))
  invoker := cluster.Join(directory)
123           

Join 函數的實作就是如下函數：

// file: cluster/cluster_impl/failover_cluster_invokers.go: newFailoverClusterInvoker()

func newFailoverClusterInvoker(directory cluster.Directory) protocol.Invoker {
  return &failoverClusterInvoker{
  baseClusterInvoker: newBaseClusterInvoker(directory),
  }
}
12345           

dubbo-go 架構預設選擇 failover 政策，既然傳回了一個 invoker，我們檢視一下 failoverClusterInvoker 的 Invoker 方法，看它是如何将叢集政策封裝到 Invoker 函數内部的：

// file: cluster/cluster_impl/failover_cluster_invokers.go: Invoker()

// Invoker 函數
func (invoker *failoverClusterInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {
  ...
  //調用List方法拿到directory緩存的所有invokers
  invokers := invoker.directory.List(invocation)
  if err := invoker.checkInvokers(invokers, invocation); err != nil {// 檢查是否可以實作調用
  return &protocol.RPCResult{Err: err}
  }
  // 擷取來自使用者方向傳入的
  methodName := invocation.MethodName()
  retries := getRetries(invokers, methodName)
  loadBalance := getLoadBalance(invokers[0], invocation)
  for i := 0; i <= retries; i++ {
  // 重要！這裡是叢集政策的展現，失敗後重試！
  //Reselect before retry to avoid a change of candidate `invokers`.
  //NOTE: if `invokers` changed, then `invoked` also lose accuracy.
  if i > 0 {
    if err := invoker.checkWhetherDestroyed(); err != nil {
    return &protocol.RPCResult{Err: err}
    }
    invokers = invoker.directory.List(invocation)
    if err := invoker.checkInvokers(invokers, invocation); err != nil {
    return &protocol.RPCResult{Err: err}
    }
  }
  // 這裡是負載均衡政策的展現！選擇特定ivk進行調用。
  ivk := invoker.doSelect(loadBalance, invocation, invokers, invoked)
  if ivk == nil {
    continue
  }
  invoked = append(invoked, ivk)
  //DO INVOKE
  result = ivk.Invoke(ctx, invocation)
  if result.Error() != nil {
    providers = append(providers, ivk.GetUrl().Key())
    continue
  }
  return result
  }
  ...
}           

看了很多 Invoke 函數的實作，所有類似的 Invoker 函數都包含兩個方向：一個是使用者方向的 invcation；一個是函數方向的底層 invokers。

而叢集政策的 invoke 函數本身作為接線員，把 invocation 一步步解析，根據調用需求和叢集政策，選擇特定的 invoker 來執行。

proxy 函數也是這樣，一個是使用者方向的 ins[] reflect.Type, 一個是函數方向的 invoker。

proxy 函數負責将 ins 轉換為 invocation，調用對應 invoker 的 invoker 函數，實作連通。

而出于這樣的設計，可以在一步步 Invoker 封裝的過程中，每個 Invoker 隻關心自己負責操作的部分，進而使整個調用棧解耦。

妙啊！！！

至此，我們了解了 failoverClusterInvoker 的 Invoke 函數實作，也正是和這個叢集政策 Invoker 被傳回，接受來自上方的調用。

已完成圖（一）中的：

5）在 zookeeper 上注冊目前 client

拿到 invokers 後，可以回到這個函數了：

// file: config/refrence_config.go: Refer()
  
  if len(c.urls) == 1 {
  // 這一步通路到registry/protocol/protocol.go registryProtocol.Refer
  c.invoker = extension.GetProtocol(c.urls[0].Protocol).Refer(*c.urls[0])
  // （一）拿到了真實的invokers
  } else {
  // 如果有多個注冊中心，即有多個invoker,則采取叢集政策
  invokers := make([]protocol.Invoker, 0, len(c.urls))
  ...
  cluster := extension.GetCluster(hitClu)
  // If 'zone-aware' policy select, the invoker wrap sequence would be:
  // ZoneAwareClusterInvoker(StaticDirectory) ->
  // FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
  c.invoker = cluster.Join(directory.NewStaticDirectory(invokers))
  }
  // （二）create proxy，為函數配置代理
  if c.Async {
  callback := GetCallback(c.id)
  c.pxy = extension.GetProxyFactory(consumerConfig.ProxyFactory).GetAsyncProxy(c.invoker, callback, cfgURL)
  } else {
  // 這裡c.invoker已經是目的addr了
  c.pxy = extension.GetProxyFactory(consumerConfig.ProxyFactory).GetProxy(c.invoker, cfgURL)
  }           

我們有了可以打通的 invokers，但還不能直接調用，因為 invoker 的入參是 invocation，而調用函數使用的是具體的參數清單，需要通過一層 proxy 來規範入參和出參。

接下來建立一個預設 proxy，放置在 c.proxy 内，以供後續使用。

至此，完成了圖（一）中最後的操作：

3. 将調用邏輯以代理函數的形式寫入 rpc-service

上面完成了 config.Refer 操作，回到：

config/config_loader.go: loadConsumerConfig()

下一個重要的函數是 Implement，它的操作較為簡單：旨在使用上面生成的 c.proxy 代理，連結使用者自己定義的 rpcService 到 clusterInvoker 的資訊傳輸。

函數較長，隻選取了重要的部分:

// file: common/proxy/proxy.go: Implement()

// Implement
// proxy implement
// In consumer, RPCService like:
//    type XxxProvider struct {
//    Yyy func(ctx context.Context, args []interface{}, rsp *Zzz) error
//    }
// Implement 實作的過程，就是proxy根據函數名和傳回值，通過調用invoker 構造出擁有遠端調用邏輯的代理函數
// 将目前rpc所有可供調用的函數注冊到proxy.rpc内
func (p *Proxy) Implement(v common.RPCService) {
  // makeDubboCallProxy 這是一個構造代理函數，這個函數的傳回值是func(in []reflect.Value) []reflect.Value 這樣一個函數
  // 這個被傳回的函數是請求實作的載體，由他來發起調用擷取結果
  makeDubboCallProxy := func(methodName string, outs []reflect.Type) func(in []reflect.Value) []reflect.Value {
  return func(in []reflect.Value) []reflect.Value {
    // 根據methodName和outs的類型，構造這樣一個函數，這個函數能将in 輸入的value轉換為輸出的value
    // 這個函數具體的實作如下：
    ...
    // 目前拿到了 methodName、所有入參的interface和value，出參數reply
    // （一）根據這些生成一個 rpcinvocation
    inv = invocation_impl.NewRPCInvocationWithOptions(
    invocation_impl.WithMethodName(methodName),
    invocation_impl.WithArguments(inIArr),
    invocation_impl.WithReply(reply.Interface()),
    invocation_impl.WithCallBack(p.callBack),
    invocation_impl.WithParameterValues(inVArr))
    for k, value := range p.attachments {
    inv.SetAttachments(k, value)
    }
    // add user setAttachment
    atm := invCtx.Value(constant.AttachmentKey) // 如果傳入的ctx裡面有attachment，也要寫入inv
    if m, ok := atm.(map[string]string); ok {
    for k, value := range m {
      inv.SetAttachments(k, value)
    }
    }
    // 至此構造inv完畢
    // (二）觸發Invoker 之前已經将cluster_invoker放入proxy，使用Invoke方法，通過getty遠端過程調用
    result := p.invoke.Invoke(invCtx, inv)
    // 如果有attachment，則加入
    if len(result.Attachments()) > 0 {
    invCtx = context.WithValue(invCtx, constant.AttachmentKey, result.Attachments())
    }
    ...
  }
  }
  numField := valueOfElem.NumField()
  for i := 0; i < numField; i++ {
  t := typeOf.Field(i)
  methodName := t.Tag.Get("dubbo")
  if methodName == "" {
    methodName = t.Name
  }
  f := valueOfElem.Field(i)
  if f.Kind() == reflect.Func && f.IsValid() && f.CanSet() { // 針對于每個函數
    outNum := t.Type.NumOut()
    // 規定函數輸出隻能有1/2個
    if outNum != 1 && outNum != 2 {
    logger.Warnf("method %s of mtype %v has wrong number of in out parameters %d; needs exactly 1/2",
      t.Name, t.Type.String(), outNum)
    continue
    }
    // The latest return type of the method must be error.
    // 規定最後一個傳回值一定是error
    if returnType := t.Type.Out(outNum - 1); returnType != typError {
    logger.Warnf("the latest return type %s of method %q is not error", returnType, t.Name)
    continue
    }
    // 擷取到所有的出參類型，放到數組裡
    var funcOuts = make([]reflect.Type, outNum)
    for i := 0; i < outNum; i++ {
    funcOuts[i] = t.Type.Out(i)
    }
    // do method proxy here:
    // （三）調用make函數，傳入函數名和傳回值，獲得能調用遠端的proxy，将這個proxy替換掉原來的函數位置
    f.Set(reflect.MakeFunc(f.Type(), makeDubboCallProxy(methodName, funcOuts)))
    logger.Debugf("set method [%s]", methodName)
  }
  }
  ...
}           

正如之前所說，proxy 的作用是将使用者定義的函數參數清單，轉化為抽象的 invocation 傳入 Invoker，進行調用。

其中已标明有三處較為重要的地方：

1. 在代理函數中實作由參數清單生成 Invocation 的邏輯
2. 在代理函數實作調用 Invoker 的邏輯
3. 将代理函數替換為原始 rpc-service 對應函數

至此，也就解決了一開始的問題：

// file: client.go: main()
  
  config.Load()
  user := &User{}
  err := userProvider.GetUser(context.TODO(), []interface{}{"A001"}, user)           

這裡直接調用使用者定義的 rpcService 的函數 GetUser，此處實際調用的是經過重寫入的函數代理，是以就能實作遠端調用了。

從 client 到 server 的 invoker 嵌套鍊- 小結

在閱讀 dubbo-go 源碼的過程中，我們能夠發現一條清晰的 invoker-proxy 嵌套鍊，希望能夠通過圖的形式來展現：

如果你有任何疑問，歡迎釘釘掃碼加入釘釘交流群：釘釘群号 23331795。

作者簡介

李志信 (GitHubID LaurenceLiZhixin)，中山大學軟體工程專業在校學生，擅長使用 Java/Go 語言，專注于雲原生和微服務等技術方向。

“ 阿裡巴巴雲原生 關注微服務、Serverless、容器、Service Mesh 等技術領域、聚焦雲原生流行技術趨勢、雲原生大規模的落地實踐，做最懂雲原生開發者的公衆号。”