一, 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封装类
<!-- server端配置-->
<motan:basicService>标签 == > BasicServiceInterfaceConfig.class封装类
<motan:service>标签 == > ServiceConfigBean.class封装类
<!-- client端配置 -->
<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进行服务引用.
