导读:随着微服务架构的流行,许多高性能 rpc 框架应运而生,由阿里开源的 dubbo 框架 go 语言版本的 dubbo-go 也成为了众多开发者不错的选择。本文将介绍 dubbo-go 框架的基本使用方法,以及从 export 调用链的角度进行 server 端源码导读,希望能引导读者进一步认识这款框架。下周将发表本文的姊妹篇:《从 client 端源码导读 dubbo-go 框架》。
近日阅读了部分dubbo-go源码 https://github.com/dubbogo/dubbo-go
当拿到一款框架之后,一种不错的源码阅读方式大致如下:从运行最基础的helloworld demo 源码开始,再查看配置文件,开启各种依赖服务(比如zk、consul),开启服务端,再到通过client调用服务端,打印完整请求日志和回包。调用成功之后,再根据框架的设计模型,从配置文件解析开始,自顶向下递阅读整个框架的调用栈。 对于C/S模式的rpc请求来说,整个调用栈被拆成了client和server两部分,所以可以分别从server端的配置文件解析阅读到server端的监听启动,从client端的配置文件解析阅读到一次invoker Call 调用。这样一次完整请求就明晰了起来。
官方demo
$ git clone https://github.com/dubbogo/dubbo-samples.git
$ cd dubbo-samples/golang/helloworld/dubbo
进入目录后可看到四个文件夹,分别支持go和java的client以及server,我们尝试运行一个go的server 进入app子文件夹内,可以看到里面保存了go文件。
$ cd go-server/app
可在go-server里面看到三个文件夹:app、assembly、profiles 其中app文件夹下保存go源码,assembly文件夹下保存可选的针对特定环境的build脚本,profiles下保存配置文件。对于dubbo-go框架,配置文件非常重要,没有文件将导致服务无法启动。
由于dubbo-go框架依赖配置文件启动,让框架定位到配置文件的方式就是通过环境变量来找。对于server端需要两个必须配置的环境变量:CONF_PROVIDER_FILE_PATH、APP_LOG_CONF_FILE,分别应该指向服务端配置文件、日志配置文件。 在sample里面,我们可以使用dev环境,即profiles/dev/log.yml profiles/dev/server.yml 两个文件。 在app/下,通过命令行中指定好这两个文件:
$ export CONF_PROVIDER_FILE_PATH="../profiles/dev/server.yml" $ export APP_LOG_CONF_FILE="../profiles/dev/log.yml"
$ go run .
如果提示timeout,则需要设置goproxy代理
$ export GOPROXY="http://goproxy.io"
再运行go run 即可开启服务
安装zookeeper,并运行zkServer, 默认为2181端口
$ cd go-client/app
$ export CONF_CONSUMER_FILE_PATH="../profiles/dev/client.yml" $ export APP_LOG_CONF_FILE="../profiles/dev/log.yml"
配置go代理
$ export GOPROXY="http://goproxy.io"
$ go run .
即可在日志中找到打印出的请求结果
response result: &{A001 Alex Stocks 18 2020-10-28 14:52:49.131 +0800 CST}
同样,在运行的server中,也可以在日志中找到打印出的请求:
req:[]interface {}{"A001"} rsp:main.User{Id:"A001", Name:"Alex Stocks", Age:18, Time:time.Time{...}
恭喜!一次基于dubbo-go的rpc调用成功。
export APP_LOG_CONF_FILE="../profiles/dev/log.yml" export CONF_CONSUMER_FILE_PATH="../profiles/dev/client.yml"
dubbo-go框架的example提供的目录如下:
源码放置在app/文件夹下,主要包含server.go 和user.go 两个文件,顾名思义,server.go用于使用框架开启服务以及注册传输协议,user.go则定义了rpc-service结构体,以及传输协议的结构。 user.go
func init() { config.SetProviderService(new(UserProvider)) // ------for hessian2------ hessian.RegisterPOJO(&User{}) } type User struct { Id string Name string Age int32 Time time.Time } type UserProvider struct { } func (u *UserProvider) GetUser(ctx context.Context, req []interface{}) (*User, error) {
可以看到,user.go中存在init函数,是服务端代码中最先被执行的部分。User为用户自定义的传输结构体,UserProvider为用户自定义的rpc_service。 包含一个rpc函数,GetUser。 当然,用户可以自定义其他的rpc功能函数。 在init函数中,调用config的SetProviderService函数,将当前rpc_service注册在框架config上。 可以查看dubbo官方文档提供的设计图
service层下面就是config层,用户服务会逐层向下注册,最终实现服务端的暴露。 rpc-service注册完毕之后,调用hessian接口注册传输结构体User。 至此init函数执行完毕 server.go
// they are necessary: // export CONF_PROVIDER_FILE_PATH="xxx" // export APP_LOG_CONF_FILE="xxx" func main() { hessian.RegisterPOJO(&User{}) config.Load() initSignal() } func initSignal() { signals := make(chan os.Signal, 1) ...
之后执行main函数 main函数中只进行了两个操作,首先使用hessian注册组件将User结构体注册(与之前略有重复),从而可以在接下来使用getty打解包。 之后调用config.Load函数,该函数位于框架config/config_loader.go内,这个函数是整个框架服务的启动点,下面会详细讲这个函数内重要的配置处理过程。执行完Load()函数之后,配置文件会读入框架,之后根据配置文件的内容,将注册的service实现到配置结构里,再调用Export暴露给特定的registry,进而开启特定的service进行对应端口的tcp监听,成功启动并且暴露服务。 最终开启信号监听initSignal()优雅地结束一个服务的启动过程。
客户端包含client.go和user.go两个文件,其中user.go与服务端完全一致,不再赘述。 client.go
// they are necessary: // export CONF_CONSUMER_FILE_PATH="xxx" // export APP_LOG_CONF_FILE="xxx" 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() }
main函数和服务端也类似,首先将传输结构注册到hessian上,再调用config.Load()函数。在下文会介绍,客户端和服务端会根据配置类型执行config.Load()中特定的函数loadConsumerConfig()和loadProviderConfig(),从而达到“开启服务”、“调用服务”的目的。 加载完配置之后,还是通过实现服务,增加函数proxy,申请registry,reloadInvoker指向服务端ip等操作,重写了客户端实例userProvider的对应函数,这时再通过调用GetUser函数,可以直接通过invoker,调用到已经开启的服务端,实现rpc过程。 下面会从server端和client端两个角度,详细讲解服务启动、registry注册、调用过程:
xxxxx
// 配置文件内容,可以参考log 和 client
config.Load()
之前设置配置,例如:func main() { hessian.RegisterPOJO(&User{}) providerConfig := config.ProviderConfig{} yaml.Unmarshal([]byte(providerConfigStr), &providerConfig) config.SetProviderConfig(providerConfig) defaultServerConfig := dubbo.GetDefaultServerConfig() dubbo.SetServerConfig(defaultServerConfig) logger.SetLoggerLevel("warn") // info,warn config.Load() select { } }
xxxxx
// 配置文件内容,可以参考log 和 client
config.Load()
之前设置配置,例如:func main() { p := config.ConsumerConfig{} yaml.Unmarshal([]byte(consumerConfigStr), &p) config.SetConsumerConfig(p) defaultClientConfig := dubbo.GetDefaultClientConfig() dubbo.SetClientConf(defaultClientConfig) logger.SetLoggerLevel("warn") // info,warn config.Load() user := &User{} err := userProvider.GetUser(context.TODO(), []interface{}{"A001"}, user) if err != nil { log.Print(err) return } log.Print(user) }
服务暴露过程涉及到多次原始rpcService的封装、暴露,网上其他文章的图感觉太过笼统,我简要的绘制了一个用户定义服务的数据流图
在加载配置之前,框架提供了很多已定义好的协议、工厂等组件,都会在对应模块init函数内注册到extension模块上,以供接下来配置文件中进行选用。 其中重要的有:
func init() { extension.SetProxyFactory("default", NewDefaultProxyFactory) }
他的作用是将原始rpc-service进行封装,形成proxy_invoker,更易于实现远程call调用,详情可见其invoke函数。
func init() { extension.SetProtocol("registry", GetProtocol) }
他负责在将invoker暴露给对应注册中心,比如zk注册中心。
func init() { extension.SetRegistry("zookeeper", newZkRegistry) }
他合并了base_resiger,负责在服务暴露过程中,将服务注册在zookeeper注册器上,从而为调用者提供调用方法。
func init() { extension.SetProtocol(DUBBO, GetProtocol) }
他负责监听对应端口,将具体的服务暴露,并启动对应的事件handler,将远程调用的event事件传递到invoker内部,调用本地invoker并获得执行结果返回。
func init() { extension.SetProtocol(FILTER, GetProtocol) }
他负责在服务暴露过程中,将代理invoker打包,通过配置好的filter形成调用链,并交付给dubbo协议进行暴露。 上述提前注册好的框架已实现的组件,在整个服务暴露调用链中都会用到,会根据配置取其所需。
服务端需要的重要配置有三个字段:services、protocols、registries profiles/dev/server.yaml:
registries : "demoZk": protocol: "zookeeper" timeout : "3s" address: "127.0.0.1:2181" services: "UserProvider": # 可以指定多个registry,使用逗号隔开;不指定默认向所有注册中心注册 registry: "demoZk" protocol : "dubbo" # 相当于dubbo.xml中的interface interface : "com.ikurento.user.UserProvider" loadbalance: "random" warmup: "100" cluster: "failover" methods: - name: "GetUser" retries: 1 loadbalance: "random" protocols: "dubbo": name: "dubbo" port: 20000
其中service指定了要暴露的rpc-service名("UserProvider),暴露的协议名("dubbo"),注册的协议名("demoZk"),暴露的服务所处的interface,负载均衡策略,集群失败策略,调用的方法等等。 其中中间服务的协议名需要和registries下的mapkey对应,暴露的协议名需要和protocols下的mapkey对应 可以看到上述例子中,使用了dubbo作为暴露协议,使用了zookeeper作为中间注册协议,并且给定了端口。如果zk需要设置用户名和密码,也可以在配置中写好。
config/config_loader.go:: Load() 在上述example的main函数中,有config.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() // service config loadProviderConfig() // init the shutdown callback GracefulShutdownInit() }
在本文中,我们重点关心loadConsumerConfig()和loadProviderConfig()两个函数 对于provider端,可以看到loadProviderConfig()函数代码如下:
前半部分是配置的读入和检查,进入for循环后,是单个service的暴露起始点。 前面提到,在配置文件中已经写好了要暴露的service的种种信息,比如服务名、interface名、method名等等。在图中for循环内,会将所有service的服务依次实现。 for循环的第一行,根据key调用GetProviderService函数,拿到注册的rpcService实例,这里对应上述提到的init函数中用户手动注册的自己实现的rpc-service实例:
这个对象也就成为了for循环中的rpcService变量,将这个对象注册通过Implement函数写到sys(ServiceConfig类型)上,设置好sys的key和协议组,最终调用了sys的Export方法。 此处对应流程图的部分:
至此,框架配置结构体已经拿到了所有service有关的配置,以及用户定义好的rpc-service实例,他触发了Export方法,旨在将自己的实例暴露出去。这是Export调用链的起始点。
config/service_config.go :: Export() 接下来进入ServiceConfig.Export()函数: 这个函数进行了一些细碎的操作,比如为不同的协议分配随机端口,如果指定了多个中心注册协议,则会将服务通过多个中心注册协议的registryProtocol暴露出去,我们只关心对于一个注册协议是如何操作的。还有一些操作比如生成调用url和注册url,用于为暴露做准备。
registryUrl是用来向中心注册组件发起注册请求的,对于zookeeper的话,会传入其ip和端口号,以及附加的用户名密码等信息 这个regUrl目前只存有注册(zk)相关信息,后续会补写入ServiceIvk,即服务调用相关信息,里面包含了方法名,参数等...
这个Register函数将服务实例注册了两次,一次是以Interface为key写入接口服务组内,一次是以interface和proto为key写入特定的一个唯一的服务。 后续会从common.Map里面取出来这个实例。
// 拿到一个proxyInvoker,这个invoker的url是传入的regUrl,这个地方将上面注册的service实例封装成了invoker // 这个GetProxyFactory返回的默认是common/proxy/proxy_factory/default.go // 这个默认工厂调用GetInvoker获得默认的proxyInvoker,保存了当前注册url invoker := extension.GetProxyFactory(providerConfig.ProxyFactory).GetInvoker(*regUrl) // 暴露出来 生成exporter,开启tcp监听 // 这里就该跳到registry/protocol/protocol.go registryProtocol 调用的Export,将当前proxyInvoker导出 exporter = c.cacheProtocol.Export(invoker)
这一步的GetProxyFactory("default")方法获取默认代理工厂,通过传入上述构造的regUrl,将url封装入代理invoker。 可以进入common/proxy/proxy_factory/default.go::ProxyInvoker.Invoke()函数里,看到对于common.Map取用为svc的部分,以及关于svc对应Method的实际调用Call的函数如下:
到这里,上面GetInvoker(*regUrl)返回的invoker即为proxy_invoker,他封装好了用户定义的rpc_service,并将具体的调用逻辑封装入了Invoke函数内。
为什么使用Proxy_invoker来调用? 我认为,通过这个proxy_invoke调用用户的功能函数,调用方式将更加抽象化,可以在代码中看到,通过ins和outs来定义入参和出参,将整个调用逻辑抽象化为invocation结构体,而将具体的函数名的选择,参数向下传递,reflect反射过程封装在invoke函数内,这样的设计更有利于之后远程调用。我认为这是dubbo Invoke调用链的设计思想。
至此,实现了图中对应的部分:
上面,我们执行到了exporter = c.cacheProtocol.Export(invoker) 这里的cacheProtocol为一层缓存设计,对应到原始的demo上,这里是默认实现好的registryProtocol registry/protocol/protocol.go:: Export() 这个函数内构造了多个EventListener,非常有java的设计感。 我们只关心服务暴露的过程,先忽略这些监听器。
一层缓存操作,如果cache没有需要从common里面重新拿zkRegistry
上述拿到了具体的zkRegistry实例,该实例的定义在 registry/zookeeper/registry.go
该结构体组合了registry.BaseRegistry结构,base结构定义了注册器基础的功能函数,比如Registry、Subscribe等,但在这些默认定义的函数内部,还是会调用facade层(zkRegistry层)的具体实现函数,这一设计模型能在保证已有功能函数不需要重复定义的同时,引入外层函数的实现,类似于结构体继承却又复用了代码。 这一设计模式我认为值得学习。 我们查看上述registry/protocol/protocol.go:: Export()函数,直接调用了:
// 1. 通过zk注册器,调用Register()函数,将已有@root@rawurl注册到zk上 err := reg.Register(*registeredProviderUrl)
将已有RegistryUrl注册到了zkRegistry上。 这一步调用了baseRegistry的Register函数,进而调用zkRegister的DoRegister函数,进而调用:
在这个函数里,将对应root创造一个新的节点
并且写入具体node信息,node为url经过encode的结果,包含了服务端的调用方式。 这部分的代码较为复杂,具体可以看baseRegistry的 processURL()函数。 至此,将服务端调用url注册到了zookeeper上,而客户端如果想获取到这个url,只需要传入特定的dubboPath,向zk请求即可。目前client是可以获取到访问方式了,但服务端的特定服务还没有启动,还没有开启特定协议端口的监听,这也是registry/protocol/protocol.go:: Export()函数接下来要做的事情:
// invoker封装入warppedInvoker wrappedInvoker := newWrappedInvoker(invoker, providerUrl) // 经过为invoker增加filter调用链,再使用dubbo协议Export,开启service并且返回了Exporter 。 // export_1 cachedExporter = extension.GetProtocol(protocolwrapper.FILTER).Export(wrappedInvoker)
新建一个WrappedInvoker,用于之后链式调用 拿到提前实现并注册好的ProtocolFilterWrapper,调用Export方法,进一步暴露。 protocol/protocolwrapped/protocol_filter_wrapper.go:Export()
protocol/protocolwrapped/protocol_filter_wrapper.go:buildInvokerChain
可见,根据配置的内容,通过链式调用的构造,层层将proxy_invoker包裹在调用链的最底部,最终返回一个调用链invoker。 对应图中部分:
至此,我们已经拿到filter调用链,期待将这个chain暴露到特定端口,用于相应请求事件
protocol/protocolwrapped/protocol_filter_wrapper.go:Export()
// 通过dubbo协议Export dubbo_protocol调用的 export_2 return pfw.protocol.Export(invoker)
回到上述Export函数的最后一行,调用了dubboProtocol的Export方法,将上述chain真正暴露。 该Export方法的具体实现在 protocol/dubbo/dubbo_protocol.go: Export()
这一函数做了两个事情:构造触发器、启动服务
将传入的Invoker调用chain进一步封装,封装成一个exporter,再将这个export放入map保存。注意!这里昂exporter放入了SetExporterMap中,在下面服务启动的时候,会以注册事件监听器的形式将这个exporter取出!
调用dubboProtocol的openServer方法,开启一个针对特定端口的监听
如上图所示,一个Session被传入,开启对应端口的事件监听。 至此构造出了exporter,完成图中部分:
上述只是启动了服务,但还没有看到触发事件的细节,点进上面的s.newSession可以看到,dubbo协议为一个getty的session默认使用了如下配置
其中很重要的一个配置是EventListener,传入的是dubboServer的默认rpcHandler protocol/dubbo/listener.go:OnMessage() rpcHandler有一个实现好的OnMessage函数,根据getty的API,当client调用该端口时,会触发OnMessage
// OnMessage notified when RPC server session got any message in connection func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) {
这一函数实现了在getty session接收到rpc 调用后的一系列处理:
整个被调过程一气呵成。实现了从getty.Session的调用事件,到经过层层封装的invoker的调用。 至此,一次rpc调用得以正确返回。
如果你有任何疑问,欢迎钉钉扫码加入交流群【钉钉群号 23331795】:
作者简介
李志信 (GitHubID LaurenceLiZhixin),中山大学软件工程专业在校学生,擅长使用 Java/Go 语言,专注于云原生和微服务等技术方向。