关于dubbo:聊聊dubbogo的nacosRegistry

序

本文次要钻研一下 dubbo-go 的 nacosRegistry

nacosRegistry

dubbo-go-v1.4.2/registry/nacos/registry.go

var (localIP = "")

const (
    //RegistryConnDelay registry connection delay
    RegistryConnDelay = 3
)

func init() {localIP, _ = gxnet.GetLocalIP()
    extension.SetRegistry(constant.NACOS_KEY, newNacosRegistry)
}

type nacosRegistry struct {
    *common.URL
    namingClient naming_client.INamingClient
}

• nacosRegistry 定义了 common.URL、namingClient 属性

newNacosRegistry

dubbo-go-v1.4.2/registry/nacos/registry.go

func newNacosRegistry(url *common.URL) (registry.Registry, error) {nacosConfig, err := getNacosConfig(url)
    if err != nil {return nil, err}
    client, err := clients.CreateNamingClient(nacosConfig)
    if err != nil {return nil, err}
    registry := nacosRegistry{
        URL:          url,
        namingClient: client,
    }
    return &registry, nil
}

• newNacosRegistry 办法先通过 getNacosConfig 获取 nacosConfig，之后通过 clients.CreateNamingClient(nacosConfig) 创立 client，最初实例化 nacosRegistry

getNacosConfig

dubbo-go-v1.4.2/registry/nacos/registry.go

func getNacosConfig(url *common.URL) (map[string]interface{}, error) {
    if url == nil {return nil, perrors.New("url is empty!")
    }
    if len(url.Location) == 0 {return nil, perrors.New("url.location is empty!")
    }
    configMap := make(map[string]interface{}, 2)

    addresses := strings.Split(url.Location, ",")
    serverConfigs := make([]nacosConstant.ServerConfig, 0, len(addresses))
    for _, addr := range addresses {ip, portStr, err := net.SplitHostPort(addr)
        if err != nil {return nil, perrors.WithMessagef(err, "split [%s]", addr)
        }
        port, _ := strconv.Atoi(portStr)
        serverConfigs = append(serverConfigs, nacosConstant.ServerConfig{
            IpAddr: ip,
            Port:   uint64(port),
        })
    }
    configMap["serverConfigs"] = serverConfigs

    var clientConfig nacosConstant.ClientConfig
    timeout, err := time.ParseDuration(url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT))
    if err != nil {return nil, err}
    clientConfig.TimeoutMs = uint64(timeout.Seconds() * 1000)
    clientConfig.ListenInterval = 2 * clientConfig.TimeoutMs
    clientConfig.CacheDir = url.GetParam(constant.NACOS_CACHE_DIR_KEY, "")
    clientConfig.LogDir = url.GetParam(constant.NACOS_LOG_DIR_KEY, "")
    clientConfig.Endpoint = url.GetParam(constant.NACOS_ENDPOINT, "")
    clientConfig.Username = url.GetParam(constant.NACOS_USERNAME, "")
    clientConfig.Password = url.GetParam(constant.NACOS_PASSWORD, "")
    clientConfig.NamespaceId = url.GetParam(constant.NACOS_NAMESPACEID, "")
    clientConfig.NotLoadCacheAtStart = true
    configMap["clientConfig"] = clientConfig

    return configMap, nil
}

• getNacosConfig 办法解析 url.Location 为 addresses，而后遍历 addresses 创立 serverConfigs，之后设置 clientConfig 的参数，最初返回 configMap

Register

dubbo-go-v1.4.2/registry/nacos/registry.go

func (nr *nacosRegistry) Register(url common.URL) error {serviceName := getServiceName(url)
    param := createRegisterParam(url, serviceName)
    isRegistry, err := nr.namingClient.RegisterInstance(param)
    if err != nil {return err}
    if !isRegistry {return perrors.New("registry [" + serviceName + "] to  nacos failed")
    }
    return nil
}

• Register 办法通过 getServiceName(url) 获取 serviceName，之后通过 createRegisterParam(url, serviceName) 创立注册参数，最初通过 nr.namingClient.RegisterInstance(param) 执行注册

Subscribe

dubbo-go-v1.4.2/registry/nacos/registry.go

//subscribe from registry
func (nr *nacosRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) {
    for {if !nr.IsAvailable() {logger.Warnf("event listener game over.")
            return
        }

        listener, err := nr.subscribe(url)
        if err != nil {if !nr.IsAvailable() {logger.Warnf("event listener game over.")
                return
            }
            logger.Warnf("getListener() = err:%v", perrors.WithStack(err))
            time.Sleep(time.Duration(RegistryConnDelay) * time.Second)
            continue
        }

        for {serviceEvent, err := listener.Next()
            if err != nil {logger.Warnf("Selector.watch() = error{%v}", perrors.WithStack(err))
                listener.Close()
                return
            }

            logger.Infof("update begin, service event: %v", serviceEvent.String())
            notifyListener.Notify(serviceEvent)
        }

    }
}

• Subscribe 办法执行 nr.subscribe(url) 获取 listener，而后通过 listener.Next() 获取 serviceEvent，而后通过 notifyListener.Notify(serviceEvent) 公布 serviceEvent

小结

newNacosRegistry 办法先通过 getNacosConfig 获取 nacosConfig，之后通过 clients.CreateNamingClient(nacosConfig) 创立 client，最初实例化 nacosRegistry

doc

• registry