Dubbo服务发布流程(服务bean的定义、创建、初始化、服务暴露、网络监听)

淡淡的烟草味﹌ 2022-05-14 19:45 307阅读 0赞

先上**简易**结论：针对一个服务发布是在底层打开socket监听端口(默认netty)，接受请求时解析请求参数(请求接口，请求方法，请求参数)去获取接口实现类的动态代理，根据反射调用实现类方法，然后(通过网络)返回请求结果给调用方

dubbo是无缝接入spring的，也因此借着spring的特性去实现一些服务发布需要做的事情(ioc实例化自定义的bean、容器初始化后进行服务发布)

1.服务发布bean的定义、初始化  
定义一个服务发布需要在xml(官方推荐用xml配置dubbo)定义下面内容：

很明显，以上<bean>标签能由spring处理后作为一个bean在容器注册，但<dubbo:service>标签是spring不能处理的，所以由dubbo"告诉"spring应该怎么处理  
dubbo有个DubboNamespaceHandler类，继承了spring的NamespaceHandlerSupport(命名空间解析)，往解析器列表中加入能解析dubbo标签的解析器

public class DubboNamespaceHandler extends NamespaceHandlerSupport { 
    
        static { 
            Version.checkDuplicate(DubboNamespaceHandler.class);
        }
    
        @Override
        public void init() { 
            registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
            registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
            registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
            registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
            registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
            registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
            registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
            registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
            registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
            registerBeanDefinitionParser("annotation", new AnnotationBeanDefinitionParser());
        }
    }

在初始化方法调用父类*NamespaceHandlerSupport*方法registerBeanDefinitionParser()，代码如下：

private final Map<String, BeanDefinitionParser> parsers = new HashMap<String, BeanDefinitionParser>();
    
    protected final void registerBeanDefinitionParser(String elementName, BeanDefinitionParser parser) { 
    		this.parsers.put(elementName, parser);
    }

所以dubbo往spring解析器列表(parsers)中加入能处理特定命名空间的dubbo标签(service、reference等)的解析器DubboBeanDefinitionParser，这样spring在读取xml解析标签时遇到dubbo的标签就会调用相应的解析器的parse（）方法去解析  
接下来查看DubboBeanDefinitionParser的构造以及重要的解析方法parse()

public class DubboBeanDefinitionParser implements BeanDefinitionParser { 
    
        private final Class<?> beanClass;
        private final boolean required;
        
    	/** * beanClass:在DubboNamespaceHandler中传进来的Class,即把标签和Class对应起来，这里例子指的是ServiceBean.class * <dubbo:service> ---解析为--- ServiceBean */
        public DubboBeanDefinitionParser(Class<?> beanClass, boolean required) { 
            this.beanClass = beanClass;
            this.required = required;
        }
        
    	@Override
        public BeanDefinition parse(Element element, ParserContext parserContext) { 
            return parse(element, parserContext, beanClass, required);
        }
    
        @SuppressWarnings("unchecked")
        private static BeanDefinition parse(Element element, ParserContext parserContext, Class<?> beanClass, boolean required) { 
            RootBeanDefinition beanDefinition = new RootBeanDefinition();
            ......
            if (ServiceBean.class.equals(beanClass)) { 
                String className = element.getAttribute("class");
                if (className != null && className.length() > 0) { 
                    RootBeanDefinition classDefinition = new RootBeanDefinition();
                    classDefinition.setBeanClass(ReflectUtils.forName(className));
                    classDefinition.setLazyInit(false);
                    parseProperties(element.getChildNodes(), classDefinition);
                    beanDefinition.getPropertyValues().addPropertyValue("ref", new BeanDefinitionHolder(classDefinition, id + "Impl"));
            }
            ......
            //把传进来的标签element里的属性(如id、interface、class等)设置给beanDefinition
            //根据beanClass(ServiceBean)获取set方法放进beanDefinition记录着
            return beanDefinition;
        }
    }

上面代码是spring初始化遇到dubbo标签时会执行的代码，这里beanDefinition的定义是记录着需要实例化bean的各种信息，相当于模子，有了模子就可以实例化相应的bean出来，返回的beanDefinition最终会放到spring一个**beanDefinitionMap**<String, BeanDefinition>中，其中**key为xml定义的id**  
spring容器初始化期间会调用如下的代码：

public class DefaultListableBeanFactory extends AbstractAutowireCapableBeanFactory
    		implements ConfigurableListableBeanFactory, BeanDefinitionRegistry, Serializable { 
    	
    	
    	@Override
    	public void preInstantiateSingletons() throws BeansException { 
    		// 实例化非懒加载的单例bean
    		
    		// 获取beanName列表
    		List<String> beanNames = new ArrayList<String>(this.beanDefinitionNames);
    
    		// 遍历beanName列表调用getBean(beanName)实例化bean
    		for (String beanName : beanNames) { 
    			RootBeanDefinition bd = getMergedLocalBeanDefinition(beanName);
    			if (!bd.isAbstract() && bd.isSingleton() && !bd.isLazyInit()) { 
    				......
    				getBean(beanName);
    				......
    			}
    		}		
    	}	
    }

当初次调用容器的getBean(beanName)时就会通过beanDefinitionMap获取BeanDefinition去实例化bean(注入依赖、在合适时机调用监听方法等)，当然这些是spring的内容了，这里就不展开了。  
经过上面步骤可以知道serviceBean是怎么来的，也许有个疑问就是 服务发布了吗？答案是没有，服务的发布是在bean创建后根据回调方法调用的export()方法，下面放serviceBean代码(只放比较重要的)：

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent>, BeanNameAware { 
    
    	/** * ApplicationListener接口方法 * 在容器初始化完成后调用 * export()这里就是服务暴露的入口 */
    	@Override
        public void onApplicationEvent(ContextRefreshedEvent event) { 
            if (isDelay() && !isExported() && !isUnexported()) { 
                if (logger.isInfoEnabled()) { 
                    logger.info("The service ready on spring started. service: " + getInterface());
                }
                export();
            }
        }
    
    	/** * InitializingBean接口方法 * 在getBean()方法实例化bean时，给bean设置完属性之后调用 * 通过applicationContext获取设置的bean(如providerBean、moduleBean等)然后赋值给自己 * * provider：其实就是看有没有配置<dubbo:provider> * application：其实就是看有没有配置<dubbo:application> * module：其实就是看有没有配置<dubbo:module> * registries：其实就是看有没有配置<dubbo:registry> * monitor：其实就是看有没有配置<dubbo:monitor> */
    	@Override
        @SuppressWarnings({ "unchecked", "deprecation"})
        public void afterPropertiesSet() throws Exception { 
            if (getProvider() == null) { 
                Map<String, ProviderConfig> providerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProviderConfig.class, false, false);
                if (providerConfigMap != null && providerConfigMap.size() > 0) { 
                	......
    	            ProviderConfig providerConfig = null;
    	            for (ProviderConfig config : providerConfigMap.values()) { 
    	                if (config.isDefault() == null || config.isDefault().booleanValue()) { 
    	                    if (providerConfig != null) { 
    	                        throw new IllegalStateException("Duplicate provider configs: " + providerConfig + " and " + config);
    	                    }
    	                    providerConfig = config;
    	                }
    	            }
    	            if (providerConfig != null) { 
    	                setProvider(providerConfig);
    	            }
                }
            }
            if (getApplication() == null
                    && (getProvider() == null || getProvider().getApplication() == null)) { 
                Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);
                if (applicationConfigMap != null && applicationConfigMap.size() > 0) { 
                    ApplicationConfig applicationConfig = null;
                    for (ApplicationConfig config : applicationConfigMap.values()) { 
                        if (config.isDefault() == null || config.isDefault().booleanValue()) { 
                            if (applicationConfig != null) { 
                                throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);
                            }
                            applicationConfig = config;
                        }
                    }
                    if (applicationConfig != null) { 
                        setApplication(applicationConfig);
                    }
                }
            }
            if (getModule() == null
                    && (getProvider() == null || getProvider().getModule() == null)) { 
                Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);
                if (moduleConfigMap != null && moduleConfigMap.size() > 0) { 
                    ModuleConfig moduleConfig = null;
                    for (ModuleConfig config : moduleConfigMap.values()) { 
                        if (config.isDefault() == null || config.isDefault().booleanValue()) { 
                            if (moduleConfig != null) { 
                                throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);
                            }
                            moduleConfig = config;
                        }
                    }
                    if (moduleConfig != null) { 
                        setModule(moduleConfig);
                    }
                }
            }
            if ((getRegistries() == null || getRegistries().isEmpty())
                    && (getProvider() == null || getProvider().getRegistries() == null || getProvider().getRegistries().isEmpty())
                    && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().isEmpty())) { 
                Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);
                if (registryConfigMap != null && registryConfigMap.size() > 0) { 
                    List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();
                    for (RegistryConfig config : registryConfigMap.values()) { 
                        if (config.isDefault() == null || config.isDefault().booleanValue()) { 
                            registryConfigs.add(config);
                        }
                    }
                    if (registryConfigs != null && !registryConfigs.isEmpty()) { 
                        super.setRegistries(registryConfigs);
                    }
                }
            }
            if (getMonitor() == null
                    && (getProvider() == null || getProvider().getMonitor() == null)
                    && (getApplication() == null || getApplication().getMonitor() == null)) { 
                Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
                if (monitorConfigMap != null && monitorConfigMap.size() > 0) { 
                    MonitorConfig monitorConfig = null;
                    for (MonitorConfig config : monitorConfigMap.values()) { 
                        if (config.isDefault() == null || config.isDefault().booleanValue()) { 
                            if (monitorConfig != null) { 
                                throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
                            }
                            monitorConfig = config;
                        }
                    }
                    if (monitorConfig != null) { 
                        setMonitor(monitorConfig);
                    }
                }
            }
            if ((getProtocols() == null || getProtocols().isEmpty())
                    && (getProvider() == null || getProvider().getProtocols() == null || getProvider().getProtocols().isEmpty())) { 
                Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false);
                if (protocolConfigMap != null && protocolConfigMap.size() > 0) { 
                    List<ProtocolConfig> protocolConfigs = new ArrayList<ProtocolConfig>();
                    for (ProtocolConfig config : protocolConfigMap.values()) { 
                        if (config.isDefault() == null || config.isDefault().booleanValue()) { 
                            protocolConfigs.add(config);
                        }
                    }
                    if (protocolConfigs != null && !protocolConfigs.isEmpty()) { 
                        super.setProtocols(protocolConfigs);
                    }
                }
            }
            if (getPath() == null || getPath().length() == 0) { 
                if (beanName != null && beanName.length() > 0
                        && getInterface() != null && getInterface().length() > 0
                        && beanName.startsWith(getInterface())) { 
                    setPath(beanName);
                }
            }
            if (!isDelay()) { 
                export();
            }
        }
    
    }

export()为父类ServiceConfig的方法，继续进入：

public class ServiceConfig<T> extends AbstractServiceConfig { 
    	
    	//判断是否延迟暴露，不是则调用doExport()
    	public synchronized void export() { 
            ......
            if (delay != null && delay > 0) { 
                delayExportExecutor.schedule(new Runnable() { 
                    @Override
                    public void run() { 
                        doExport();
                    }
                }, delay, TimeUnit.MILLISECONDS);
            } else { 
                doExport();
            }
        }
    	
    	protected synchronized void doExport() { 
    		//状态改为"已暴露"
    		//属性初始化操作(application、protocols、provider、module、registries等)，从System.getProperty获取默认值
    		//判断属性是否符合预期
            ......
            doExportUrls();//暴露服务
            ......
        }
        
        /** * 获取注册url列表 * 获取暴露服务时用的协议列表(默认dubbo协议暴露服务) */
    	@SuppressWarnings({ "unchecked", "rawtypes"})
        private void doExportUrls() { 
            List<URL> registryURLs = loadRegistries(true);
            for (ProtocolConfig protocolConfig : protocols) { 
                doExportUrlsFor1Protocol(protocolConfig, registryURLs);
            }
        }
    	
    	/** * 1.构建暴露服务用的URL * 2.根据url.scope判断远程暴露还是本地暴露 * * @param protocolConfig 协议配置 * @param registryURLs 注册url */
    	private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) { 
            String name = protocolConfig.getName();//协议名称
            if (name == null || name.length() == 0) { 
                name = "dubbo";
            }
            Map<String, String> map = new HashMap<String, String>();//参数
            //给map配置参数(methods、application、interface、revision等等)
            ......
            ......
            
            String contextPath = protocolConfig.getContextpath();//服务治理
            String host = this.findConfigedHosts(protocolConfig, registryURLs, map);//暴露服务的IP地址
            Integer port = this.findConfigedPorts(protocolConfig, name, map);//暴露服务端口
    		//最终就是为了构建一个暴露服务用的url
    		URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
    		//例如dubbo://192.168.41.21:21881/com.zc.selectCourse.service.CoursesService?anyhost=true&application=euu-new-service-provider-consumer&bind.ip=192.168.41.21&bind.port=21881&dubbo=2.6.2&generic=false&interface=com.zc.selectCourse.service.CoursesService&methods=findCount,coursesList,findByCourses,save,update,getRequiredList,importCourse,findAll,isIntersectionDjj2,collectDelete,isIntersectionDjj,findStudentsByXH,isCoursesHave,findByCode,isCanSelect,findByCourseClassifyId,findById,deleteById,collectList,collect&pid=8092&revision=1.0&side=provider&timestamp=1540798928292
    		
    		//如果不在xml指定scope属性则为null，既会本地暴露又会远程暴露
    		String scope = url.getParameter(Constants.SCOPE_KEY);
            if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) { 
                // 判断是否本地暴露
                if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) { 
                    exportLocal(url);
                }
                // 判断是否远程暴露
                if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) { 
                	......
                    for (URL registryURL : registryURLs) { 
                        //根据要暴露的具体实现类ref(即Impl)，暴露接口，注册url去生成invoker(封装了实现类以便于服务访问的统一规范)
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                        DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);//再把invoke封装一层
    					
    					//暴露服务，打开底层网络端口监听请求
                        Exporter<?> exporter = protocol.export(wrapperInvoker);
                        exporters.add(exporter);
                    }
                    ......
                }
            }
            this.urls.add(url);
        }
    }

从ServiceBean作为入口到其父类doExportUrlsFor1Protocol()方法去暴露服务，其调用顺序如下：

ServiceBeanServiceConfigonApplicationEvent(ContextRefreshedEvent event)export()doExport()doExportUrls()doExportUrlsFor1Protocol(protocolConfig, registryURLs)getInvoker(ref,interfaceClass)和export(wrapperInvoker)ServiceBeanServiceConfig

然后说一下proxyFactory的由来

private static final ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

proxyFactory是dubbo利用了SPI的思想去构造出来(可以理解根据接口和配置文件获取实现类)，返回的是动态生成的类ProxyFactory$Adaptive，调用getInvoker()会根据调用链依次调用

生成封装着ref的AbstractProxyInvoker返回

ProxyFactory$Adaptive.getInvoker(T, Class, URL)

StubProxyFactoryWrapper.getInvoker(T, Class, URL)

(重要)JavassistProxyFactory.getInvoker(T, Class, URL)

结束

下面看JavassistProxyFactory的getInvoker方法：

public class JavassistProxyFactory extends AbstractProxyFactory { 
    	
    	/** * 客户端调用的方法，返回接口代理类，代理内容就是通过网络去请求服务端的实现类 */
        @Override
        @SuppressWarnings("unchecked")
        public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) { 
            return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
        }
    	
    	/** * 服务端调用方法，根据实现类去获取invoker，用于暴露服务 */
        @Override
        public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) { 
    		//生成一个动态代理类wrapper，代理Impl实现类，通过反射的方法去调用实现类的方法
            final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
            //直接new一个invoker，封装这实现类ref
            //调用invoker.doinvoke()方法就是通过反射去调用ref的方法
            return new AbstractProxyInvoker<T>(proxy, type, url) { 
                @Override
                protected Object doInvoke(T proxy, String methodName,
                                          Class<?>[] parameterTypes,
                                          Object[] arguments) throws Throwable { 
                    return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
                }
            };
        }
    
    }

总而言之：getInvoker()就是返回一个封装这实现类的invoker，这个invoker可以调用doInvoke()方法去调用实现类的方法，问为什么要生成invoke返回，直接返回ref不就行了吗----答：远程调用规范，全部暴露都以invoke为参数，统一通过反射调用实现类方法  
getInvoker()方法执行完之后就是服务的暴露了

private static final Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
    
    //暴露服务，打开底层网络端口监听请求
    Exporter<?> exporter = protocol.export(wrapperInvoker);

和上面同理，protocol的生成也是通过SPI原理生成的Protocol$Adaptive，export(invoker)方法的调用链是

向注册中心注册url

生成DubboExporter,打开socket监听

Protocol$Adaptive.export(invoker)

QosProtocolWrapper.export(invoker)

ProtocolListenerWrapper.export(invoker)

ProtocolFilterWrapper.export(invoker)

RegistryProtocol.export(invoker)

(重要)DubboProtocol.export(invoker)

结束

…  
//未完待续