2023年6月25日发(作者：)

dubbo@reference为null_dubbo--服务发现源码分析⼀. 起源Spring IOC中bean的⽣成分为两个阶段，⼀个是实例化阶段，另⼀个是初始化阶段。实例化阶段是⽣成将class变成对象的过程，以及参与⼀部分成员赋值，在这⼀过程中，可以通过InstantiationAwareBeanPostProcessor进⾏⼲预处理。另⼀个阶段主要是赋值，调⽤初始化⽅法等，在这⼀阶段，可以通过BeanPostProcessor对Bean做⼀些操作，例如@ Autowired和@Resource等就是这⼀阶段处理的。当⼀个Bean初始化时，如果这个Bean中有@Reference（即需要注⼊dubbo的代理类）,dubbo就会调⽤ReferenceAnnotationBeanPostProcessor对Bean的⽣成进⾏⼲预，ReferenceAnnotationBeanPostProcessor继承了InstantiationAwareBeanPostProcessorAdapter，是InstantiationAwareBeanPostProcessor的⼦⼦⼦类，Sping会调⽤该处理器的postProcessPropertyValues⽅法，⽅法的⼊参包含Bean的信息，再通过inject⽅法注⼊：@Overridepublic PropertyValues postProcessPropertyValues( PropertyValues pvs, PropertyDescriptor[] pds, Object bean, String beanName) throws BeanCreationException { //⼊参中包含了已经实例化号的bean，和当前bean的名字 InjectionMetadata metadata = findReferenceMetadata(beanName, ss(), pvs); try { (bean, beanName, pvs); } catch (BeanCreationException ex) { throw ex; } catch (Throwable ex) { throw new BeanCreationException(beanName, "Injection of @Reference dependencies failed", ex); } return pvs;}然后通过层层调⽤到达buildReferenceBean⽅法，在这个⽅法⾥开始构建ReferenceBean，ReferenceBean是服务发现的核⼼类：private ReferenceBean buildReferenceBean(nce reference, Class referenceClass) throws Exception { String referenceBeanCacheKey = generateReferenceBeanCacheKey(reference, referenceClass); ReferenceBean referenceBean = (referenceBeanCacheKey); if (referenceBean == null) { ReferenceBeanBuilder beanBuilder = ReferenceBeanBuilder .create(reference, classLoader, applicationContext) .interfaceClass(referenceClass); //build模式进⾏构建 referenceBean = (); bsent(referenceBeanCacheKey, referenceBean); } return referenceBean;}build⽅法的代码：public final B build() throws Exception { checkDependencies(); //duBuild()⽅法⾥⾯new了⼀个ReferenceBean对象 B bean = doBuild(); //对ReferenceBean进⾏配置调⽤ configureBean(bean); if (Enabled()) { (bean + " has been built."); } return bean;}configureBean⽅法的代码：protected void configureBean(B bean) throws Exception { preConfigureBean(annotation, bean); configureRegistryConfigs(bean); configureMonitorConfig(bean); configureApplicationConfig(bean); configureModuleConfig(bean); postConfigureBean(annotation, bean);}configureBean就是为ReferenceBean做⼀些参数配置，重点在最后⼀⾏的postConfigureBean⽅法：@Overrideprotected void postConfigureBean(nce annotation, ReferenceBean bean) throws Exception { licationContext(applicationContext); configureInterface(annotation, bean); configureConsumerConfig(annotation, bean); ropertiesSet();}postConfigureBean的前三⾏任然是对ReferenceBean的参数，最后⼀⾏调⽤了ReferenceBean的afterPropertiesSet的⽅法，正式开启服务发现。ReferenceBean实现了FactoryBean，ApplicationContextAware，InitializingBean和DisposableBean接⼝，这些接⼝都是SpringIOC对Bean管理所使⽤的接⼝，实际上ReferenceBean的实例化对象并不归Spring IOC管，⼀个ReferenceBean对应⼀个dubbo服务类，Spring也管不了，ReferenceBean的接⼝逻辑都是dubbo代码⽀撑起的实现，⽽不是Spring，dubbo源码中对ReferenceBean的实例化对象也叫bean，不清楚的⼈很容易就理解为Spring所管理的，特此说明。⼆. #afterPropertiesSet开启服务发现afterPropertiesSet⽅法主要是想ApplicationContext拿取配置，如果ApplicationContext中没有，就使⽤默认的，然后调⽤getObject()⽅法进⼀步处理，再经过get()⽅法最终调⽤init()⽅法，init()⽅法会⽣成ReferenceBean中变量名为ref的成员变量，ref这个成员变量正是@Re ference注解注⼊的动态代理对象，所以ref的⽣成就⼗分重要，下⾯是init()⽅法源码：/** * ref初始化，这个⽅法分为两段。 * 第⼀段不难发现map是⼀个配置容器，这段代码的⼤部分操作都是向map中添加配置 * 第⼆段则是根据map配置调⽤createProxy⽅法创建ref，ref也是最终被注⼊的动态代理类 */private void init() { if (initialized) { return; } initialized = true; if (interfaceName == null || () == 0) { throw new IllegalStateException(" interface not allow null!"); } // get consumer's global configuration checkDefault(); appendProperties(this); if (getGeneric() == null && getConsumer() != null) { setGeneric(getConsumer().getGeneric()); } if (ric(getGeneric())) { interfaceClass = ; } else { try { interfaceClass = e(interfaceName, true, tThread() .getContextClassLoader()); } catch (ClassNotFoundException e) { throw new IllegalStateException(sage(), e); } checkInterfaceAndMethods(interfaceClass, methods); } String resolve = perty(interfaceName); String resolveFile = null; if (resolve == null || () == 0) { resolveFile = perty(""); if (resolveFile == null || () == 0) { File userResolveFile = new File(new File(perty("")), "ties"); if (()) { resolveFile = olutePath(); } } if (resolveFile != null && () > 0) { Properties properties = new Properties(); FileInputStream fis = null; try { fis = new FileInputStream(new File(resolveFile)); (fis); } catch (IOException e) { throw new IllegalStateException("Unload " + resolveFile + ", cause: " + sage(), e); } finally { try { if (null != fis) (); } catch (IOException e) { (sage(), e); } } resolve = perty(interfaceName); } } if (resolve != null && () > 0) { url = resolve; if (Enabled()) { if (resolveFile != null) { ("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service."); } else { ("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service."); } } } if (consumer != null) { if (application == null) { application = lication(); } if (module == null) { module = ule(); } if (registries == null) { registries = istries(); } if (monitor == null) { monitor = itor(); } } if (module != null) { if (registries == null) { registries = istries(); } if (monitor == null) { monitor = itor(); } } if (application != null) { if (registries == null) { registries = istries(); } if (monitor == null) { monitor = itor(); } } checkApplication(); checkStubAndMock(interfaceClass); Map map = new HashMap(); Map attributes = new HashMap(); (_KEY, ER_SIDE); (_VERSION_KEY, tocolVersion()); (AMP_KEY, f(tTimeMillis())); if (() > 0) { (_KEY, f(())); } if (!isGeneric()) { String revision = sion(interfaceClass, version); if (revision != null && () > 0) { ("revision", revision); } String[] methods = pper(interfaceClass).getMethodNames(); if ( == 0) { ("NO method found in service interface " + e()); ("methods", _VALUE); } else { ("methods", (new HashSet((methods)), ",")); } } (ACE_KEY, interfaceName); appendParameters(map, application); appendParameters(map, module); appendParameters(map, consumer, T_KEY); appendParameters(map, this); String prefix = viceKey(map); if (methods != null && !y()) { for (MethodConfig method : methods) { appendParameters(map, method, e()); String retryKey = e() + ".retry"; if (nsKey(retryKey)) { String retryValue = (retryKey); if ("false".equals(retryValue)) { (e() + ".retries", "0"); } } appendAttributes(attributes, method, prefix + "." + e()); checkAndConvertImplicitConfig(method, map, attributes); } } String hostToRegistry = temProperty(_IP_TO_REGISTRY); if (hostToRegistry == null || () == 0) { hostToRegistry = alHost(); } else if (isInvalidLocalHost(hostToRegistry)) { throw new IllegalArgumentException("Specified invalid registry ip from property:" + _IP_TO_REGISTRY + ", value:" + hostToRegistry); } (ER_IP_KEY, hostToRegistry); //attributes are stored by system context. temContext().putAll(attributes); ref = createProxy(map); ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, hods()); nsumerModel(getUniqueServiceName(), consumerModel);}上⾯的代码⾮常长，但是逻辑⼗分清晰，这段代码可分为两段，第⼀段不难发现map是⼀个配置容器，这段代码的⼤部分操作都是向map中添加配置，第⼆段则是使⽤这个配置map调⽤createProxy⽅法创建代理类，继续深⼊createProxy⽅法：private T createProxy(Map map) { URL tmpUrl = new URL("temp", "localhost", 0, map); final boolean isJvmRefer; //判断是不是在jvm中引⽤ if (isInjvm() == null) { if (url != null && () > 0) { // if a url is specified, don't do local reference isJvmRefer = false; } else if (vmProtocol().isInjvmRefer(tmpUrl)) { // by default, reference local service if there is isJvmRefer = true; } else { isJvmRefer = false; } } else { isJvmRefer = isInjvm().booleanValue(); }

//如果是在jvm中引⽤，⽣成对应的url，并由refprotocol⽣成Invoker，refprotocol是⼀个动态代理类，能通过不同的url来实现不同的逻辑 if (isJvmRefer) { URL url = new URL(_PROTOCOL, OST, 0, e()).addParameters(map); invoker = (interfaceClass, url); if (Enabled()) { ("Using injvm service " + e()); } } else { //如果⽤户指定了URL if (url != null && () > 0) { // user specified URL, could be peer-to-peer address, or register center's address. String[] us = LON_SPLIT_(url); if (us != null && > 0) { for (String u : us) { URL url = f(u); if (h() == null || h().length() == 0) { url = h(interfaceName); } if (RY_(tocol())) { (ameterAndEncoded(_KEY, yString(map))); } else { (rl(url, map)); } } } //如果没有指定 URL，就去加载注册中⼼的URL } else { // assemble URL from register center's configuration List us = loadRegistries(false); if (us != null && !y()) { if (us != null && !y()) { //加载注册中⼼的监控 for (URL u : us) { URL monitorUrl = loadMonitor(u); if (monitorUrl != null) { (R_KEY, (String())); } (ameterAndEncoded(_KEY, yString(map))); } } if (y()) { } } //如果注册中⼼只有⼀个，refprotocol直接⽣成Invoker if (() == 1) { invoker = (interfaceClass, (0)); } else { //如果注册中⼼有多个，则调⽤cluster的join⽅法来⽣成Invoker List> invokers = new ArrayList>(); URL registryURL = null; for (URL url : urls) { ((interfaceClass, url)); if (RY_(tocol())) { registryURL = url; // use last registry url } } if (registryURL != null) { // registry url is available // use AvailableCluster only when register's cluster is available URL u = ameter(R_KEY, ); invoker = (new StaticDirectory(u, invokers)); } else { // not a registry url invoker = (new StaticDirectory(invokers)); } } }

//检查 Boolean c = check; if (c == null && consumer != null) { c = k(); } if (c == null) { c = true; // default true } //是否可⽤ if (c && !lable()) { } if (Enabled()) { ("Refer dubbo service " + e() + " from url " + ()); } // create service proxy //最终调⽤proxyFactory的getProxy⽅法⽣成最终的代理类 return (T) xy(invoker);} throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + alHost() + " use dubbo ve throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" :

createProxy的⽅法稍微复杂⼀点，⾸先判断该引⽤是不是jvm中的，如果是就直接创建Invoker，其次判断⽤户是否指定的URL，如果未指定，就加载注册中⼼的URL，这⼀阶段的主要⽬的是拿到URL，然后再判断URL是不是有多个，如果只有⼀个，就调⽤refprotocol创建Invoker，如果有多个，就调⽤cluster的join⽅法合并多个Invoker为⼀个Invoker，然后检查，最终调⽤proxyFactory的getProxy⽅法⽣成ref，可以明显看出，⼤部分逻辑都是如何⽣成Invoker，这⾥⾯有⼏个细节需要继续深⼊，⼀个是refprotocol的ref⽅法，另⼀个是cluster的join⽅法，最后⼀个是proxyFactory的getProxy⽅法。流程图如下：服务发现流程图1. 深⼊refprotocol#ref由于ReferenceBean中引⼊的refprotocol是由SPI机制⽣成的动态代理类，想要看到具体的代码，有两种办法，第⼀种是断点ExtensionLoader的createAdaptiveExtensionClassCode⽅法，查看⽣成的code，第⼆种是⽤Arthas反编译，这⾥选⽤第⼆种。启动Arthas，先⽤sc搜索对应的class：红框中的类就是想要反编译的，然后再⽤命令jad ol$Adaptive进⾏反编译，得到的代码如下：public class Protocol$Adaptiveimplements Protocol { @Override public void destroy() { } @Override public int getDefaultPort() { } public Exporter export(Invoker invoker) throws RpcException { String string; if (invoker == null) { throw new IllegalArgumentException("r argument == null"); } if (() == null) { throw new IllegalArgumentException("r argument getUrl() == null"); } URL uRL = (); String string2 = string = tocol() == null ? "dubbo" : tocol(); if (string == null) { } Protocol protocol = ensionLoader().getExtension(string); return (invoker); } public Invoker refer(Class class_, URL uRL) throws RpcException { String string; if (uRL == null) { throw new IllegalArgumentException("url == null"); } URL uRL2 = uRL; //获取协议，如果为空，则默认为dubbo String string2 = string = tocol() == null ? "dubbo" : tocol(); if (string == null) { } Protocol protocol = ensionLoader().getExtension(string); return (class_, uRL); }} throw new UnsupportedOperationException("method public abstract void y() of interface throw new UnsupportedOperationException("method public abstract int aultPort() of interface . throw new IllegalStateException(new StringBuffer().append("Fail to get extension(ol) name from url(").append(ng( throw new IllegalStateException(new StringBuffer().append("Fail to get extension(ol) name from url(").append(ng重点关注refer⽅法，这个⽅法从url中取协议，如果协议为空，则默认为dubbo，然后⽤ExtensionLoader加载指定的拓展，然后再次debug这个getExtension⽅法，发现得到的Protocol是⼀个套娃结构，以zookeeper为注册中⼼时的registry为例：最外层是ProtocolFilterWrapper，然后是QosProtocolWrapper，然后是ProtocolListennerWrapper，最后才是RegistryProtocol。ProtocolFilterWrapper的作⽤是啥呢？我们在服务调动的时候可以有Filter对服务进⾏拦截处理，例如sofa-tracer对服务调⽤时，⽣产者和消费者打印⽇志中的trace_id⼀致，就是⽤的拦截器对消费者和⽣成都做了拦截，下⾯是ProtocolFilterWrapper的源码：private static Invoker buildInvokerChain(final Invoker invoker, String key, String group) { Invoker last = invoker; List filters = ensionLoader().getActivateExtension((), key, group); if (!y()) { for (int i = () - 1; i >= 0; i--) { final Filter filter = (i); final Invoker next = last; last = new Invoker() { @Override public Class getInterface() { return erface(); } @Override public URL getUrl() { return (); } @Override public boolean isAvailable() { return lable(); } @Override public Result invoke(Invocation invocation) throws RpcException { return (next, invocation); } @Override public void destroy() { y(); } @Override public String toString() { return ng(); } }; } } return last; }QosProtocolWrapper是什么呢，QosProtocolWrapper是⽤来启动QosServer的，Qos全程Quality of Service，Qos详细的资料可以看这⾥，QosProtocolWrapper源码如下：private void startQosServer(URL url) { if (!eAndSet(false, true)) { return; } try { boolean qosEnable = oolean(ameter(QOS_ENABLE,"true")); if (!qosEnable) { return; } int port = nt(ameter(QOS_PORT,"22222")); boolean acceptForeignIp = oolean(ameter(ACCEPT_FOREIGN_IP,"true")); Server server = tance(); t(port); eptForeignIp(acceptForeignIp); (); } catch (Throwable throwable) { //throw new RpcException("fail to start qos server", throwable); } }ProtocolListennerWrapper⼜是⼲嘛的呢？在服务运⾏的过程中，有时候需要监听某个服务，在它暴露或发现它做⼀些事情，ProtocolListennerWrapper就是⽤来实现这个功能的，源码如下，当服务被暴露或发现服务时，会通知所有的listeners： @Override public Invoker refer(Class type, URL url) throws RpcException { if (RY_(tocol())) { return (type, url); } return new ListenerInvokerWrapper((type, url), fiableList( ensionLoader() .getActivateExtension(url, R_LISTENER_KEY))); } public ListenerInvokerWrapper(Invoker invoker, List listeners) { if (invoker == null) { throw new IllegalArgumentException("invoker == null"); } r = invoker; ers = listeners; if (listeners != null && !y()) { for (InvokerListener listener : listeners) { if (listener != null) { try { ed(invoker); } catch (Throwable t) { (sage(), t); } } } } }最后终于终于到了RegistryProtocol，程序运⾏到RegistryProtocol的refer⽅法，获取对应的注册中⼼，然后调⽤⽅法doRefer，在doRefer⽅法中new了⼀个RegistryDirectory对象，然后使⽤cluster#join⽅法⽣成Invoker，还记得有多个注册中⼼URL的时候怎么处理的吗，都是调⽤的这个⽅法，在下⾯我们深⼊cluster的join⽅法，下⾯是doRefer源码：private Invoker doRefer(Cluster cluster, Registry registry, Class type, URL url) { //实例化⼀个RegistryDirectory对象 RegistryDirectory directory = new RegistryDirectory(type, url); //分别设置注册中⼼和协议 istry(registry); tocol(protocol); // all attributes of REFER_KEY Map parameters = new HashMap(().getParameters()); URL subscribeUrl = new URL(ER_PROTOCOL, (ER_IP_KEY), 0, e(), parameters); if (!_(viceInterface()) && ameter(ER_KEY, true)) { //注册中⼼注册消费者 er(ameters(RY_KEY, ERS_CATEGORY, _KEY, f(false))); } //订阅服务 ibe(ameter(RY_KEY, ERS_CATEGORY + "," + URATORS_CATEGORY + "," + S_CATEGORY)); //⽣成Invoker Invoker invoker = (directory); //放⼊缓存 erConsumer(invoker, url, subscribeUrl, directory); return invoker;}2. 深⼊cluster#joincluster也是⼀个Adaptive的动态代理类，再次通过Arthas反编译获取起代码如下：/* * Decompiled with CFR. */package r;import ;import ionLoader;import r;import eption;import r;import ory;public class Cluster$Adaptiveimplements Cluster { public Invoker join(Directory directory) throws RpcException { if (directory == null) { throw new IllegalArgumentException("ory argument == null"); } if (() == null) { throw new IllegalArgumentException("ory argument getUrl() == null"); } URL uRL = (); //获取cluster参数，默认为failover String string = ameter("cluster", "failover"); if (string == null) { } Cluster cluster = ensionLoader().getExtension(string); return (directory); }} throw new IllegalStateException(new StringBuffer().append("Fail to get extension(r) name from url(").append(可以得知该动态代理类先是获取url中cluster的值，如果为空，则默认为failover，然后再通过getExtension获取具体的Cluster对象，在此通过断点获知Cluster也是个套娃:外层是⼀个MockClusterWrapper，然后再是FailoverCluster。MockClusterWrapper中将其包装成MockClusterInvoker：public class MockClusterWrapper implements Cluster { private Cluster cluster; public MockClusterWrapper(Cluster cluster) { r = cluster; } @Override public Invoker join(Directory directory) throws RpcException { return new MockClusterInvoker(directory, (directory)); }}FailoverCluster中将其包装成FailoverClusterInvokerpublic class FailoverCluster implements Cluster { public final static String NAME = "failover"; @Override public Invoker join(Directory directory) throws RpcException { return new FailoverClusterInvoker(directory); }}也就是说最终得到⼀个这样的数据结构，MockClusterWrapper对象中有directory和FailoverClusterInvoker对象两个成员变量，⽽FailoverClusterInvoker中有成员变量directory。3. 深⼊proxyFactory#getProxy再次通过Arthas反编译proxyFactory#getProxy得到源码如下：public Object getProxy(Invoker invoker) throws RpcException { if (invoker == null) { throw new IllegalArgumentException("r argument == null"); } if (() == null) { throw new IllegalArgumentException("r argument getUrl() == null"); } URL uRL = (); String string = ameter("proxy", "javassist"); if (string == null) { } ProxyFactory proxyFactory = ensionLoader().getExtension(string); return xy(invoker); } throw new IllegalStateException(new StringBuffer().append("Fail to get extension(actory) name from url(").append(再次断点getExtension，⼜得到⼀个套娃结构：进⼊StubProxyFactoryWrapper，getProxy⽅法如下：@Override@SuppressWarnings({"unchecked", "rawtypes"})public T getProxy(Invoker invoker) throws RpcException { //通过JavassistProxyFactory获得动态代理类 T proxy = xy(invoker); if ( != erface()) { //获取存根参数 String stub = ().getParameter(_KEY, ().getParameter(_KEY)); //如果存根参数不为空 if (mpty(stub)) { Class serviceType = erface(); if (ult(stub)) { if (().hasParameter(_KEY)) { stub = e() + "Stub"; } else { stub = e() + "Local"; } } try { //获取存根class Class stubClass = e(stub); if (!gnableFrom(stubClass)) { throw new IllegalStateException("The stub implementation class " + e() + " not implement interface " + e()); } try { Constructor constructor = nstructor(stubClass, serviceType); //实例化class proxy = (T) tance(new Object[]{proxy}); //export stub service //导出存根服务 URL url = (); if (ameter(_EVENT_KEY, T_STUB_EVENT)) { url = ameter(_SERVER_KEY, ng()); try { export(proxy, (Class) erface(), url); } catch (Exception e) { ("export a stub service error.", e); } } } catch (NoSuchMethodException e) { } } catch (Throwable t) { // ignore } } } return proxy;} url = ameter(_EVENT_METHODS_KEY, (pper(ss()).getDeclaredMethodNam throw new IllegalStateException("No such constructor "public " + pleName() + "(" + e() + ")" in stub implementa ("Failed to create stub implementation class " + stub + " in consumer " + alHost() + " use dubbo version " + 这段代码的逻辑是使⽤JavassistProxyFactory⽣成代理类，然后检查URL中有没有“stub”参数。如果存在，就导出存根服务。JavassistProxyFactory创建代理的⽅法如下： @Override @SuppressWarnings("unchecked") public T getProxy(Invoker invoker, Class[] interfaces) { return (T) xy(interfaces).newInstance(new InvokerInvocationHandler(invoker)); }可知，创建爱你代理时⼜包了⼀层InvokerInvocationHandler，InvokerInvocationHandler源码如下：public class InvokerInvocationHandler implements InvocationHandler { private final Invoker invoker; public InvokerInvocationHandler(Invoker handler) { r = handler; } @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { String methodName = e(); Class[] parameterTypes = ameterTypes(); if (laringClass() == ) { return (invoker, args); } if ("toString".equals(methodName) && == 0) { return ng(); } if ("hashCode".equals(methodName) && == 0) { return de(); } if ("equals".equals(methodName) && == 1) { return (args[0]); } //封装成RpcInvocation return (new RpcInvocation(method, args)).recreate(); }}注意invoke⽅法，被动态代理的类的⽅法都会交给invoke⽅法来处理，InvokerInvocationHandler这个类将除Object外的⽅法都封装成RpcInvocation交给invoker来调⽤。现在屡⼀下思路，⾸先服务暴露时，先⽣成URL，再根据URL来⽣成Directory，再根据cluster的injoin⽅法⽣成Invoker，然后再根据proxyFactory⽣成适配的代理类，⽽当我们调⽤dubbo服务的⽅法时，真正⼲活的是Invoker，现在反推Invoker⼀共封了多少层：1.⾸先是创建动态代理类时封的⼀层InvokerInvocationHandler2. 然后是cluster的injoin封的MockClusterInvoker和FailoverClusterInvokerinvoker的初步封装⼀共三层三.总结服务发现的过程就是通过各种途径创建Directory，然后通过cluster创建成Invoker，然后⽣成动态代理类，当调⽤⼀个dubbo服务时，实际上就是层层调⽤Invoker的⽅法，下⼀篇深⼊Invoker远程调⽤的过程。