轻量级Rpc框架设计--motan源码解析六:client端服务发现

一, Client端初始化工作

client端通过RefererConfigBean类实现InitializingBean接口的afterPropertiesSet方法, 进行下面三项检查配置工作:

①checkAndConfigBasicConfig(); // 检查并配置basicConfig

②checkAndConfigProtocols(); //检查并配置protocols

③checkAndConfigRegistry(); //检查并配置registry

1.1 motan框架client端对xml配置文件解析

motan框架在client端对xml配置文件的解析, 与对server端xml配置文件中的<motan:registry>注册中心标签和<motan:protocol>通信协议标签解析是一样的, 有区别的就是server端对于服务暴露的通用基础配置标签是<motan:basicService>, 服务暴露的标签是<motan:service>, 而client端对于服务引用的通用基础配置标签是<motan:basicReferer>, 服务引用的标签是<motan:referer>, 最终motan框架通过解析spring xml文件的配置, 将配置文件中的每个标签都对应转换成其封装类, 下面列举出motan框架部分标签对应的封装类, 如下:

<motan:registry>标签 == > 对应RegistryConfig.class封装类

<motan:protocol>标签 == > 对应ProtocolConfig.class封装类

<motan:basicService>标签 == > BasicServiceInterfaceConfig.class封装类

<motan:service>标签 == > ServiceConfigBean.class封装类

<motan:basicReferer>标签 == > BasicRefererInterfaceConfig封装类

<motan:referer>标签 == >RefererConfigBean封装类

所以, client端使用motan框架时, motan框架的spring-support模块会对client端的配置文件内容进行解析, 将不同的标签转换成其对应的封装类对象, 并且以id或者name(当id不存在时)为key, 放入Spring容器.

motan框架对server端xml配置文件解析, 可查看以前的内容: 轻量级Rpc框架设计--motan源码解析二:自定义spring标签与解析

1.2 checkAndConfigBasicConfig;// 检查并配置basicConfig

<!-- 通用referer基础配置 -->
<motan:basicReferer requestTimeout="200" accessLog="false" retries="2" group="motan-demo-rpc" module="motan-demo-rpc"application="myMotanDemo" protocol="motan" registry="registry" id="motantestClientBasicConfig" throwException="false" check="true" />

/*** 检查并配置basicConfig*/private void checkAndConfigBasicConfig() {if (getBasicReferer() == null) {for (String name : MotanNamespaceHandler.basicRefererConfigDefineNames) {BasicRefererInterfaceConfig biConfig = beanFactory.getBean(name, BasicRefererInterfaceConfig.class);if (biConfig == null) {continue;}if (MotanNamespaceHandler.basicRefererConfigDefineNames.size() == 1) {setBasicReferer(biConfig);} else if (biConfig.isDefault() != null && biConfig.isDefault().booleanValue()) {setBasicReferer(biConfig);}}}}

checkAndConfigBasicConfig()方法, 首先判断<motan:basicReferer>标签对应的BasicRefererInterfaceConfig.class封装类对象是否为空, 即判断client端配置文件中没有配置<motan:basicReferer>该标签. 否则从MotanNamespaceHandler.basicRefererConfigDefineNames全局变量中再次获取<motan:basicReferer>标签对应的name值, 有的话, 就从spring容器根据name获取BasicRefererInterfaceConfig.class封装类.

1.3 checkAndConfigProtocols(); //检查并配置protocols

/*** 检查是否已经装配protocols，否则按basicConfig--->default路径查找*/private void checkAndConfigProtocols() {if (CollectionUtil.isEmpty(getProtocols()) && getBasicReferer() != null&& !CollectionUtil.isEmpty(getBasicReferer().getProtocols())) {setProtocols(getBasicReferer().getProtocols());}if (CollectionUtil.isEmpty(getProtocols())) {for (String name : MotanNamespaceHandler.protocolDefineNames) {ProtocolConfig pc = beanFactory.getBean(name, ProtocolConfig.class);if (pc == null) {continue;}if (MotanNamespaceHandler.protocolDefineNames.size() == 1) {setProtocol(pc);} else if (pc.isDefault() != null && pc.isDefault().booleanValue()) {setProtocol(pc);}}}if (CollectionUtil.isEmpty(getProtocols())) {setProtocol(MotanFrameworkUtil.getDefaultProtocolConfig());}}

<!-- motan协议配置 --><motan:protocol default="true" name="motan" haStrategy="failover"loadbalance="roundrobin" maxClientConnection="10" minClientConnection="2"/>

checkAndConfigProtocols();方法, 首先检查client端xml配置文件中, 是否配置了<motan:protocol>, 是的话, 就使用该配置, 否则的话, 判断getBasicReferer() != null而且!CollectionUtil.isEmpty(getBasicReferer().getProtocols())也不能为空, 则就使用<motan:basicReferer>标签中对protocol通信协议的配置.

1.4 checkAndConfigRegistry(); //检查并配置registry

public void checkAndConfigRegistry() {if (CollectionUtil.isEmpty(getRegistries()) && getBasicReferer() != null&& !CollectionUtil.isEmpty(getBasicReferer().getRegistries())) {setRegistries(getBasicReferer().getRegistries());}if (CollectionUtil.isEmpty(getRegistries())) {for (String name : MotanNamespaceHandler.registryDefineNames) {RegistryConfig rc = beanFactory.getBean(name, RegistryConfig.class);if (rc == null) {continue;}if (MotanNamespaceHandler.registryDefineNames.size() == 1) {setRegistry(rc);} else if (rc.isDefault() != null && rc.isDefault().booleanValue()) {setRegistry(rc);}}}if (CollectionUtil.isEmpty(getRegistries())) {setRegistry(MotanFrameworkUtil.getDefaultRegistryConfig());}}

<motan:registry regProtocol="zookeeper" name="registry" address="127.0.0.1:2181" requestTimeout="5000"/>

检查并配置registry, 首先判断client端xml配置文件中没有配置<motan:registry>的话, 判断getBasicReferer() != null不能为空, 而且<motan:basicReferer>公共基础标签中配置了注册中心相关信息, 则!CollectionUtil.isEmpty(getBasicReferer().getRegistries())不为空.

二, client客户端服务引用的过程

2.1 通过getBean获取服务的代理对象时, 相当于FactoryBean#getObject()代理了getBean()方法。

MotanDemoService service = (MotanDemoService) ctx.getBean("motanDemoReferer");

而RefererConfigBean实现了FactoryBean接口的getObject()方法, 如下:

public class RefererConfigBean<T> extends RefererConfig<T> implements FactoryBean<T>, BeanFactoryAware, InitializingBean, DisposableBean {private static final long serialVersionUID = 8381310907161365567L;private transient BeanFactory beanFactory;@Overridepublic T getObject() throws Exception {return getRef();}

2.2 getRef()方法分析

public T getRef() {if (ref == null) {initRef();}return ref;}

public synchronized void initRef() {if (initialized.get()) {return;}try {interfaceClass = (Class) Class.forName(interfaceClass.getName(), true, Thread.currentThread().getContextClassLoader());} catch (ClassNotFoundException e) {throw new MotanFrameworkException("ReferereConfig initRef Error: Class not found " + interfaceClass.getName(), e,MotanErrorMsgConstant.FRAMEWORK_INIT_ERROR);}if (CollectionUtil.isEmpty(protocols)) {throw new MotanFrameworkException(String.format("%s RefererConfig is malformed, for protocol not set correctly!",interfaceClass.getName()));}checkInterfaceAndMethods(interfaceClass, methods);clusterSupports = new ArrayList<ClusterSupport<T>>(protocols.size());List<Cluster<T>> clusters = new ArrayList<Cluster<T>>(protocols.size());String proxy = null;// 获取SimpleConfigHandlerConfigHandler configHandler = ExtensionLoader.getExtensionLoader(ConfigHandler.class).getExtension(MotanConstants.DEFAULT_VALUE);// 载入注册的配置, 并且转换成URL对象// [zookeeper://127.0.0.1:2181/com.weibo.api.motan.registry.RegistryService?group=default_rpc]List<URL> registryUrls = loadRegistryUrls();// 获取当前clien端的所在地址信息, 192.168.99.1String localIp = getLocalHostAddress(registryUrls);for (ProtocolConfig protocol : protocols) {Map<String, String> params = new HashMap<String, String>();params.put(URLParamType.nodeType.getName(), MotanConstants.NODE_TYPE_REFERER);params.put(URLParamType.version.getName(), URLParamType.version.getValue());params.put(URLParamType.refreshTimestamp.getName(), String.valueOf(System.currentTimeMillis()));// 从protocol, basicReferer, extConfig, this(即RefererConfig)这些对象中收集collect配置信息, 并且写入params参数collectConfigParams(params, protocol, basicReferer, extConfig, this);// 类似于dubbo框架对于client配置服务引用时, 提供方法级别的配置, 比如超时设置等,如下:// <dubbo:service> <dubbo:method name="" timeout="10000" retries="9" /><dubbo:service/> // 而motan框架也提供了client端引用服务时, 对方法级别上的配置, 如下:// <motan:referer><motan:method name="" retries=""/></motan:referer>, // collectMethodConfigParams()方法就是收集,client端服务引用时方法级别上的配置.collectMethodConfigParams(params, this.getMethods());URL refUrl = new URL(protocol.getName(), localIp, MotanConstants.DEFAULT_INT_VALUE, interfaceClass.getName(), params);// 根据服务引用配置信息refUrl, 处理对象configHandler, 注册中心registryUrls, client客户端创建集群支持, // 比如设置haStrategy高可用策略, loadBalance负载均衡策略等ClusterSupport<T> clusterSupport = createClusterSupport(refUrl, configHandler, registryUrls);clusterSupports.add(clusterSupport);clusters.add(clusterSupport.getCluster());proxy = (proxy == null) ? refUrl.getParameter(URLParamType.proxy.getName(), URLParamType.proxy.getValue()) : proxy;}ref = configHandler.refer(interfaceClass, clusters, proxy);initialized.set(true);}

①首先根据client客户端xml配置文件中对服务引用时, 配置的接口全限定名称加载该接口,

interfaceClass = (Class) Class.forName(interfaceClass.getName(), true, Thread.currentThread().getContextClassLoader());

②如果client端配置服务引用时, 同时也在方法级别上做了配置, 则此处的checkInterfaceAndMethods(interfaceClass, methods);就是进行校验该接口是否包含配置的这些方法, 进行合法性校验的

checkInterfaceAndMethods(interfaceClass, methods);

③获取SimplConfigHandler, 根据配置转换URL对象, 获取当前client客户端节点所在的地址信息, 用于以后注册写到zk上.

// 获取SimpleConfigHandler
ConfigHandler configHandler = ExtensionLoader.getExtensionLoader(ConfigHandler.class).getExtension(MotanConstants.DEFAULT_VALUE);// 载入注册的配置, 并且转换成URL对象
// [zookeeper://127.0.0.1:2181/com.weibo.api.motan.registry.RegistryService?group=default_rpc]
List<URL> registryUrls = loadRegistryUrls();// 获取当前clien端的所在地址信息, 192.168.99.1
String localIp = getLocalHostAddress(registryUrls);

④从protocol, basicReferer, extConfig, this(即RefererConfig)这些对象中collect收集配置信息, 并且写入params参数

 // 从protocol, basicReferer, extConfig, this(即RefererConfig)这些对象中收集collect配置信息, 并且写入params参数
collectConfigParams(params, protocol, basicReferer, extConfig, this);

⑤收集client端服务引用时, 方法级别上的配置信息:

// 类似于dubbo框架对于client配置服务引用时, 提供方法级别的配置, 比如超时设置等,如下:
// <dubbo:service> <dubbo:method name="" timeout="10000" retries="9" /><dubbo:service/>
// 而motan框架也提供了client端引用服务时, 对方法级别上的配置, 如下:
// <motan:referer><motan:method name="" retries=""/></motan:referer>,
// collectMethodConfigParams()方法就是收集,client端服务引用时方法级别上的配置.
collectMethodConfigParams(params, this.getMethods());

类似于dubbo框架对于client配置服务引用时, 提供方法级别的配置, 比如超时设置等,如下: <dubbo:service> <dubbo:method name="" timeout="10000" retries="9" /><dubbo:service/>
而motan框架也提供了client端引用服务时, 对方法级别上的配置, 如下:<motan:referer><motan:method name="" retries=""/></motan:referer>,

collectMethodConfigParams()方法就是收集,client端引用服务时在方法级别上的配置.

⑥关键代码, 初始化cluster集群环境, 比如, haStrategy高可用策略, loadBalance负载均衡策略等

// 根据服务引用配置信息refUrl, 处理对象configHandler, 注册中心registryUrls, 初始化cluster集群,
// 比如初始化cluster集群的haStrategy高可用策略, loadBalance负载均衡策略等
ClusterSupport<T> clusterSupport = createClusterSupport(refUrl, configHandler, registryUrls);

2.3 关键代码分析, 初始化cluster集群, createClusterSupport()方法.

private ClusterSupport<T> createClusterSupport(URL refUrl, ConfigHandler configHandler, List<URL> registryUrls) {List<URL> regUrls = new ArrayList<URL>();// 如果用户指定directUrls 或者 injvm协议访问，则使用local registryif (StringUtils.isNotBlank(directUrl) || MotanConstants.PROTOCOL_INJVM.equals(refUrl.getProtocol())) {URL regUrl =new URL(MotanConstants.REGISTRY_PROTOCOL_LOCAL, NetUtils.LOCALHOST, MotanConstants.DEFAULT_INT_VALUE,RegistryService.class.getName());if (StringUtils.isNotBlank(directUrl)) {StringBuilder duBuf = new StringBuilder(128);String[] dus = MotanConstants.COMMA_SPLIT_PATTERN.split(directUrl);for (String du : dus) {if (du.contains(":")) {String[] hostPort = du.split(":");URL durl = refUrl.createCopy();durl.setHost(hostPort[0].trim());durl.setPort(Integer.parseInt(hostPort[1].trim()));durl.addParameter(URLParamType.nodeType.getName(), MotanConstants.NODE_TYPE_SERVICE);duBuf.append(StringTools.urlDecode(durl.toFullStr())).append(MotanConstants.COMMA_SEPARATOR);}}if (duBuf.length() > 0) {duBuf.deleteCharAt(duBuf.length() - 1);regUrl.addParameter(URLParamType.directUrl.getName(), duBuf.toString());}}regUrls.add(regUrl);} else { // 通过注册中心配置拼装URL，注册中心可能在本地，也可能在远端if (registryUrls == null || registryUrls.isEmpty()) {throw new IllegalStateException(String.format("No registry to reference %s on the consumer %s , please config <motan:registry address=\"...\" /> in your spring config.",interfaceClass, NetUtils.LOCALHOST));}for (URL url : registryUrls) {regUrls.add(url.createCopy());}}for (URL url : regUrls) {url.addParameter(URLParamType.embed.getName(), StringTools.urlEncode(refUrl.toFullStr()));}return configHandler.buildClusterSupport(interfaceClass, regUrls);}

首先,判断如果用户指定directUrls 或者 injvm协议访问，则使用local registry, 否则通过注册中心配置拼装URL,根据interfaceClass服务引用接口,注册中心的配置信息, SimpleConfigHandler类调用buildClusterSupport()方法, 初始化cluster集群.

①SimpleConfigHandler类初始化集群的buildClusterSupport()方法分析

public <T> ClusterSupport<T> buildClusterSupport(Class<T> interfaceClass, List<URL> registryUrls) {ClusterSupport<T> clusterSupport = new ClusterSupport<T>(interfaceClass, registryUrls);// init()方法, 实际上完成了三件事// 1 初始化cluster集群// 2 向zk注册// 3 订阅server服务变化通知clusterSupport.init();return clusterSupport;}

clusterSupport.init()方法总的来说, 完成了两件事, 一是, 初始化cluster集群, prepareCluster()方法, 当client客户端没有配置haStrategy和loadBalance策略时, 使用motan框架默认的策略, 二是, client端向zk注册自己(位置:/motan/group/xxxx/client/ip/). 三是, 通过watcher机制, 实现订阅server服务变化通知(服务下线, 配置变更), 并且会调用NotifyListener进行集群的刷新操作.

②clusterSupport.init()方法分析,

public void init() {// 准备集群环境, 即判断xml配置文件中是否配置hastrategy和loadbalance策略, // 没有配置的话, motan框架就使用自己默认的策略prepareCluster();URL subUrl = toSubscribeUrl(url);for (URL ru : registryUrls) {String directUrlStr = ru.getParameter(URLParamType.directUrl.getName());// 如果有directUrl，直接使用这些directUrls进行初始化，不用到注册中心discoverif (StringUtils.isNotBlank(directUrlStr)) {List<URL> directUrls = parseDirectUrls(directUrlStr);if (!directUrls.isEmpty()) {notify(ru, directUrls);LoggerUtil.info("Use direct urls, refUrl={}, directUrls={}", url, directUrls);continue;}}// 获取操作zk的客户端连接对象Registry registry = getRegistry(ru);// subscribe()方法完成了两件事, 一是, client端在zk的/motan/group/接口全限定名称/client/client节点ip, 注册自己(即写操作)// 二是, client端通过watch机制实现服务的订阅功能, 当服务发生变更(服务下线, 配置变更)时, client端就会调用NotifyListener进行集群的刷新操作.registry.subscribe(subUrl, this);}boolean check = Boolean.parseBoolean(url.getParameter(URLParamType.check.getName(), URLParamType.check.getValue()));if (!CollectionUtil.isEmpty(cluster.getReferers()) || !check) {cluster.init();if (CollectionUtil.isEmpty(cluster.getReferers()) && !check) {LoggerUtil.warn(String.format("refer:%s", this.url.getPath() + "/" + this.url.getVersion()), "No services");}return;}throw new MotanFrameworkException(String.format("ClusterSupport No service urls for the refer:%s, registries:%s",this.url.getIdentity(), registryUrls), MotanErrorMsgConstant.SERVICE_UNFOUND);}

首先准备集群环境prepareCluster(), 即判断xml配置文件中是否配置hastrategy和loadbalance策略,没有配置的话, motan框架就使用自己默认的策略. 接着获取操作zk的客户端连接对象getRegistry(ru);最后调用subscribe()方法完成两件事, 一是, client端在zk的/motan/group/接口全限定名称/client/client节点ip, 注册自己(即写操作), 二是, client端通过watch机制实现服务的订阅功能, 当服务发生变更(服务下线, 配置变更)时, client端就会调用NotifyListener进行集群的刷新操作.

③prepareCluster();方法, 配置集群环境, 如下:

首先从url对象获取xml配置文件中的负载均衡策略, 没有的话就默认使用activeWeight(低并发度优先： referer的某时刻的call数越小优先级越高).当然我们这里已经配置了haStrategy和loadbalance策略, 分别为:faileover(故障转移)和roundrobin(轮询), 所以motan框架就会使用这俩分别作为高可用策略和负载均衡策略.

<motan:protocol default="true" name="motan" haStrategy="failover"
loadbalance="roundrobin" maxClientConnection="10" minClientConnection="2"/>

private void prepareCluster() {String clusterName = url.getParameter(URLParamType.cluster.getName(), URLParamType.cluster.getValue());// 首先从url对象获取xml配置文件中的负载均衡策略, 没有的话就默认使用activeWeight// 当然我们这里已经配置了<motan:protocol default="true" name="motan" haStrategy="failover"// loadbalance="roundrobin" maxClientConnection="10" minClientConnection="2"/>String loadbalanceName = url.getParameter(URLParamType.loadbalance.getName(), URLParamType.loadbalance.getValue());String haStrategyName = url.getParameter(URLParamType.haStrategy.getName(), URLParamType.haStrategy.getValue());cluster = ExtensionLoader.getExtensionLoader(Cluster.class).getExtension(clusterName);LoadBalance<T> loadBalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(loadbalanceName);HaStrategy<T> ha = ExtensionLoader.getExtensionLoader(HaStrategy.class).getExtension(haStrategyName);cluster.setLoadBalance(loadBalance);cluster.setHaStrategy(ha);cluster.setUrl(url);}

④getRegistry();获取操作zk的客户端连接对象

protected Registry getRegistry(URL url) {RegistryFactory registryFactory = ExtensionLoader.getExtensionLoader(RegistryFactory.class).getExtension(url.getProtocol());return registryFactory.getRegistry(url);}

⑤subscribe()方法分析

public void subscribe(URL url, NotifyListener listener) {....doSubscribe(url.createCopy(), listener);}
public void doSubscribe(URL url, NotifyListener listener) {....try {concreteSubscribe(url, listener);} catch (Exception e) {....}}
protected void concreteSubscribe(final URL url, final NotifyListener notifyListener) {....// 监听到zk上server服务发生变化时listenerIZkChildListener zkChildListener = childChangeListeners.get(notifyListener);if (zkChildListener == null) {childChangeListeners.putIfAbsent(notifyListener, new IZkChildListener() {@Overridepublic void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {// 处理方法ZookeeperRegistry.this.notify(url, notifyListener, nodeChildsToUrls(parentPath, currentChilds));LoggerUtil.info(String.format("[ZookeeperRegistry] service list change: path=%s, currentChilds=%s", parentPath, currentChilds.toString()));}});zkChildListener = childChangeListeners.get(notifyListener);}// 写入Client结点....// /motan/group/xxxx/client/ip/// 创建临时节点zkClient.createEphemeral(clientNodePath, info);// toServerTypePath(url)方法将, motan://192.168.99.1:0/com.weibo.motan.demo.service.MotanDemoService?group=motan-demo-rpc// 转换成zk注册地址, /motan/motan-demo-rpc/com.weibo.motan.demo.service.MotanDemoService/server// 并且使用zkChildListener的handleChildChange()方法处理zk节点变化通知List<String> currentChilds = zkClient.subscribeChildChanges(toServerTypePath(url), zkChildListener);....// 触发一次Notify(), 通知ClusterSupport进行一次refreshCluster()操作notify(url, notifyListener, nodeChildsToUrls(toServerTypePath(url), currentChilds));}

subscribe()方法最主要完成了三件事, 一是, 创建监听到server节点变化时的处理方法zkChildListener#handleChildChange(). 二是, client端向zk注册自己zkClient.createEphemeral(). 三是, 监听server服务节点/motan/group/xxx/server/ip/下的subscribeChildChanges事件, 并交由zkChildListener进行处理监听到的事件.四是, 触发一次Notify()方法, 通知ClusterSupport进行一次refreshCluster()操作.

⑥ClusterSupport#notify() --> refreshCluster()方法 , 刷新集群

protected void concreteSubscribe(final URL url, final NotifyListener notifyListener) {....// 触发一次Notify(), 通知ClusterSupport进行一次refreshCluster()操作notify(url, notifyListener, nodeChildsToUrls(toServerTypePath(url), currentChilds));
}protected void notify(URL refUrl, NotifyListener listener, List<URL> urls) {....for (List<URL> us : nodeTypeUrlsInRs.values()) {listener.notify(getUrl(), us);}}public synchronized void notify(URL registryUrl, List<URL> urls) {....// 判断urls中是否包含权重信息，并通知loadbalance。processWeights(urls);....// 此处不销毁referers，由cluster进行销毁registryReferers.put(registryUrl, newReferers);refreshCluster();}
// 刷新provider集群
private void refreshCluster() {List<Referer<T>> referers = new ArrayList<Referer<T>>();for (List<Referer<T>> refs : registryReferers.values()) {referers.addAll(refs);}cluster.onRefresh(referers);}

ClusterSupport实现了NotifyListener该接口的notify()方法,从而能够感知到zk中有哪些已经注册的服务, 并且回调notify()方法.

综上, 经过上面的步骤后, motan框架的client客户端就已经实现了向zk注册(创建临时节点:/motan/group/xxx/client/ip), 监听服务变化(服务下线, 配置变更), notify()触发一次refreshCluster()刷新集群操作获取provider暴露的服务.
所以, 目前客户端就能就能监听provider端在zk上暴露服务的变化状态.

2.4 configHandler.refer()采用Jdk Proxy动态代理技术远程调用一个服务

 public <T> T refer(Class<T> interfaceClass, List<Cluster<T>> clusters, String proxyType) {ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getExtension(proxyType);return proxyFactory.getProxy(interfaceClass, new RefererInvocationHandler<T>(interfaceClass, clusters));}

当client调用代理类中的方法时, 就会首先调用代理类的invoke()方法, 如下:

①client端首先调用代理类的hello()方法时, 会首先执行代理类的invoke()方法, 发起一次远程调用.

// client端调用代理类的hello()方法
for(int i = 0; i < Integer.MAX_VALUE; i++){System.out.println(service.hello("motan" + i));Thread.sleep(500);}

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {DefaultRequest request = new DefaultRequest();....// 当 referer配置多个protocol的时候,比如A,B,C, // 降级 : 那么正常情况下只会使用A，如果A被开关降级，那么就会使用B，B也被降级，那么会使用Cfor (Cluster<T> cluster : clusters) {....try {response = cluster.call(request);return response.getValue();} catch (RuntimeException e) {....}}

核心代码: cluster.call()

②cluster.call()分析.

//1 public class ClusterSpi<T>#call()
public Response call(Request request) {if (available.get()) {try {return haStrategy.call(request, loadBalance);} catch (Exception e) {return callFalse(request, e);}}....}
//2 public class FailoverHaStrategy<T>#call()
public Response call(Request request, LoadBalance<T> loadBalance) {// 首先根据lb的策略, 选择referers引用List<Referer<T>> referers = selectReferers(request, loadBalance);....return refer.call(request);....}
// 根据不同的loadbalance负载策略, 选择一个或者是一批referers
protected List<Referer<T>> selectReferers(Request request, LoadBalance<T> loadBalance) {List<Referer<T>> referers = referersHolder.get();referers.clear();loadBalance.selectToHolder(request, referers);return referers;}
//3 public class DefaultRpcProtocol{// 内部类class DefaultRpcReferer<T> extends AbstractReferer<T>{// 最终调用doCall()方法protected Response doCall(Request request) {....// 采用NettyClient调用request()进行远程服务调用return client.request(request);....}}
}

cluster.call()方法调用历经三步最终使用NettyClient#request()方法进行远程服务调用, ClusterSpi#call() --> FailoverHaStrategy#call() --> AbstractReferer#call() --> 最终调用DefaultRpcReferer#doCall()方法完成远程服务调用, 并且返回远程服务的执行结果.

为了便于理解, 在网上找到了两个相当好的有关client客户端初始化以及方法调用的说明图, 作为对以上初始化cluster集群, 方法调用的一个总结, 如下:

motan框架对于集群的管理, 图如下:

三, 总结

本片分析了motan框架client客户端引用provider暴露的服务时, 历经了哪些过程. 首先按照client配置文件中的内容初始化一个cluster集群, 接着向zk写入client端节点信息(/motan/group/com.weibo.motan.demo.service.MotanDemoService/clinet/ip/), 利用zookeeper的watcher机制, 监听provider端在zk暴露服务时, 注册的节点信息以及其所有的ChildZnode节点 (/motan/group/com.weibo.motan.demo.service.MotanDemoService/server/ip/), 当发生服务变更(比如服务下线, 配置变更)时, motan框架在client端就会调用NotifyListener#notify()方法refreshCluster()刷新provider进行服务暴露的集群.最后就是client端进行服务引用, 通过NettyClient调用远程服务, 得到执行结果.

好了, 到这里就简单的分析了motan框架provider服务注册与发布, consumer进行服务引用.

轻量级Rpc框架设计--motan源码解析六:client端服务发现相关推荐

支撑微博千亿调用的轻量级RPC框架：Motan
随着微博容器化部署以及混合云平台的高速发展,RPC 在微服务化的进程中越来越重要,对 RPC 的需求也产生了一些变化.今天主要介绍一下微博 RPC 框架 Motan,以及为了更好的适应混合云部署所做的 ...
Dubbo架构设计与源码解析（一）架构设计
作者:黄金一.架构演变单应用架构 ----> 垂直架构 ----> 分布式架构 ----> 微服务架构 ----> 云原生架构二.Dubbo总体架构 1.角色职能 • C ...
Java生鲜电商平台-秒杀系统微服务架构设计与源码解析实战
Java生鲜电商平台-秒杀系统微服务架构设计与源码解析实战 Java生鲜电商平台- 什么是秒杀通俗一点讲就是网络商家为促销等目的组织的网上限时抢购活动比如说京东秒杀,就是一种定时定量秒杀,在规定 ...
Java生鲜电商平台-促销系统的架构设计与源码解析
Java生鲜电商平台-促销系统的架构设计与源码解析说明:本文重点讲解现在流行的促销方案以及源码解析,让大家对促销,纳新有一个深入的了解与学习过程. 促销系统是电商系统另外一个比较大,也是比较复杂的系 ...
Java生鲜电商平台-电商会员体系系统的架构设计与源码解析
Java生鲜电商平台-电商会员体系系统的架构设计与源码解析说明:Java生鲜电商平台中会员体系作为电商平台的基础设施,重要性不容忽视.我去年整理过生鲜电商中的会员系统,但是比较粗,现在做一个最好的整 ...
Java生鲜电商平台-商品中心的架构设计与源码解析(小程序/APP)
Java生鲜电商平台-商品中心的架构设计与源码解析(小程序/APP) 说明:Java生鲜电商平台中,由于商品的架构很大程度决定了电商的扩展性与伸缩性.对此根据自己多年的生鲜电商经验,整理了以下的商品中 ...
Celery 源码解析六：Events 的实现
序列文章: Celery 源码解析一:Worker 启动流程概述 Celery 源码解析二:Worker 的执行引擎 Celery 源码解析三: Task 对象的实现 Celery 源码解析四: 定时 ...
Dubbo 框架设计与源码解读（配置解析优先级、线程分配、负载均衡、容错方案）
整体框架设计图例说明: 图中左边淡蓝背景的为服务消费⽅使⽤的接⼝,右边淡绿⾊背景的为服务提供⽅使⽤的接⼝,位于中轴线上的为双⽅都⽤到的接⼝. 图中从下⾄上分为⼗层,各层均为单向依赖,右边的⿊⾊箭头代 ...
从（新浪）motan看RPC框架设计
kris的文章开始计算机科学领域的任何问题都可以通过增加一个间接的中间层来解决从零开发一款RPC框架,说难也难说简单也简单.难的是你的设计将如何面对实际中的复杂应用场景:简单的是其思想可以仅仅浓缩 ...

轻量级Rpc框架设计--motan源码解析六:client端服务发现

轻量级Rpc框架设计--motan源码解析六:client端服务发现相关推荐

最新文章

热门文章