額,dubbo的前世今生咱就不說了,有用的人都知道。(最新版本已更名 incubator-dubbo,預計版本3.0,不過咱們就先不摻和別人的開發過程了,以v2.6.3為藍本)
首先,是工作需要,用到了dubbo用於微服務化各系統。所以,總體來說,是被動學習。
也就是先有了做,依葫蘆畫瓢,然后才來理解為什么!
當然,工作需要也不能全部不知道,自然是去官網看看有什么文檔,看下其工作原理,可能出現的問題。以及在工作中解決問題用到的辦法。總之,已經懂得七七八八了,但是,我們需要一個深度的學習。接下來就來走走看看!
拋個磚:dubbo提供者服務示例, 其結構是這樣的!
dubbo://192.168.11.6:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=12720&side=provider×tamp=1534902103892
拿來一個官網可以運行起來的實例,讓代碼跑起來:
// 提供者: public class Provider { public static void main(String[] args) throws Exception { //Prevent to get IPV6 address,this way only work in debug mode //But you can pass use -Djava.net.preferIPv4Stack=true,then it work well whether in debug mode or not System.setProperty("java.net.preferIPv4Stack", "true"); ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-provider.xml"}); context.start(); // 這個比較巧妙, 只需停留在當前點,不按下enter鍵,服務就會一直存在,等待消費者連接,而且無需真正提供一個監聽服務 System.in.read(); // press any key to exit } } // 消費者: public class Consumer { public static void main(String[] args) { //Prevent to get IPV6 address,this way only work in debug mode //But you can pass use -Djava.net.preferIPv4Stack=true,then it work well whether in debug mode or not System.setProperty("java.net.preferIPv4Stack", "true"); ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"}); context.start(); DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxy String hello = demoService.sayHello("world"); // call remote method, 調用遠程服務和本地服務一樣,這是其優勢所在 System.out.println(hello); // get result } } }
// xml 配置,因為使用組播協議,所以demo拿來就可以運行
// 配置很簡單,主要說明下這里使用組播地址的好處。 組播是個很好的概念,可以使網絡更快,也更方便開發,可以參考下: multicast,秒懂單播、廣播或多播及其應用場景
provider:
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:dubbo="http://dubbo.apache.org/schema/dubbo" xmlns="http://www.springframework.org/schema/beans" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsd http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd"> <!-- provider's application name, used for tracing dependency relationship --> <dubbo:application name="demo-provider"/> <!-- use multicast registry center to export service --> <dubbo:registry address="multicast://224.5.6.7:1234"/> <!-- use dubbo protocol to export service on port 20880 --> <dubbo:protocol name="dubbo" port="20880"/> <!-- service implementation, as same as regular local bean --> <bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/> <!-- declare the service interface to be exported --> <dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService"/> </beans>
consumer:

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:dubbo="http://dubbo.apache.org/schema/dubbo" xmlns="http://www.springframework.org/schema/beans" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsd http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd"> <!-- consumer's application name, used for tracing dependency relationship (not a matching criterion), don't set it same as provider --> <dubbo:application name="demo-consumer"/> <!-- use multicast registry center to discover service --> <dubbo:registry address="multicast://224.5.6.7:1234"/> <!-- generate proxy for the remote service, then demoService can be used in the same way as the local regular interface --> <dubbo:reference id="demoService" check="false" interface="com.alibaba.dubbo.demo.DemoService"/> </beans>
下面來講講其大概加載過程吧!!
// 由spring加載上下文,並到publishEvent上時,調用dubbo的event,dubbo開始介入 // spring 加載 Context public ClassPathXmlApplicationContext(String[] configLocations, boolean refresh, ApplicationContext parent) throws BeansException { super(parent); setConfigLocations(configLocations); if (refresh) { // 加載核心代碼 refresh(); } } // spring 最核心的加載代碼結構如下,在 最后一步加載 dubbo @Override public void refresh() throws BeansException, IllegalStateException { synchronized (this.startupShutdownMonitor) { // Prepare this context for refreshing. prepareRefresh(); // Tell the subclass to refresh the internal bean factory. ConfigurableListableBeanFactory beanFactory = obtainFreshBeanFactory(); // Prepare the bean factory for use in this context. prepareBeanFactory(beanFactory); try { // Allows post-processing of the bean factory in context subclasses. postProcessBeanFactory(beanFactory); // Invoke factory processors registered as beans in the context. invokeBeanFactoryPostProcessors(beanFactory); // Register bean processors that intercept bean creation. registerBeanPostProcessors(beanFactory); // Initialize message source for this context. initMessageSource(); // Initialize event multicaster for this context. initApplicationEventMulticaster(); // Initialize other special beans in specific context subclasses. onRefresh(); // Check for listener beans and register them. registerListeners(); // 完成bean的加載初始化,進行bean的創建 // Instantiate all remaining (non-lazy-init) singletons. finishBeanFactoryInitialization(beanFactory); // 在此處使用publishEvent,加載dubbo服務 // Last step: publish corresponding event. finishRefresh(); } catch (BeansException ex) { // 如果初始化異常,則先銷毀各種bean,以回收內存,清理線程等 if (logger.isWarnEnabled()) { logger.warn("Exception encountered during context initialization - " + "cancelling refresh attempt: " + ex); } // Destroy already created singletons to avoid dangling resources. destroyBeans(); // Reset 'active' flag. cancelRefresh(ex); // Propagate exception to caller. throw ex; } finally { // Reset common introspection caches in Spring's core, since we // might not ever need metadata for singleton beans anymore... resetCommonCaches(); } } }
// 進行bean的創建
// DefaultListableBeanFactory.preInstantiateSingletons() -> AbstractBeanFactory.doGetBean() -> DefaultSgetSingletonBeanFactory.getSingleton -> AbstractBeanFactory.createBean(),
// DefaultListableBeanFactory.registerBeanDefinition() 進行bean注冊
// 會有5個beanName,由DubboBeanDefinitionParser.parse()創建,demo-provider, dubbo, com.alibaba.demo.DemoService, com.alibaba.dubbo.config.RegistryConfig, demoService

// DefaultListableBeanFactory.preInstantiateSingletons() -> AbstractBeanFactory.doGetBean() -> DefaultSgetSingletonBeanFactory.getSingleton -> AbstractBeanFactory.createBean(), // DefaultListableBeanFactory.registerBeanDefinition() 進行bean注冊 // 會有5個beanName,由DubboBeanDefinitionParser.parse()創建,demo-provider, dubbo, com.alibaba.demo.DemoService, com.alibaba.dubbo.config.RegistryConfig, demoService @Override protected Object createBean(String beanName, RootBeanDefinition mbd, Object[] args) throws BeanCreationException { if (logger.isDebugEnabled()) { logger.debug("Creating instance of bean '" + beanName + "'"); } RootBeanDefinition mbdToUse = mbd; // Make sure bean class is actually resolved at this point, and // clone the bean definition in case of a dynamically resolved Class // which cannot be stored in the shared merged bean definition. Class<?> resolvedClass = resolveBeanClass(mbd, beanName); if (resolvedClass != null && !mbd.hasBeanClass() && mbd.getBeanClassName() != null) { mbdToUse = new RootBeanDefinition(mbd); mbdToUse.setBeanClass(resolvedClass); } // Prepare method overrides. try { mbdToUse.prepareMethodOverrides(); } catch (BeanDefinitionValidationException ex) { throw new BeanDefinitionStoreException(mbdToUse.getResourceDescription(), beanName, "Validation of method overrides failed", ex); } try { // Give BeanPostProcessors a chance to return a proxy instead of the target bean instance. Object bean = resolveBeforeInstantiation(beanName, mbdToUse); if (bean != null) { return bean; } } catch (Throwable ex) { throw new BeanCreationException(mbdToUse.getResourceDescription(), beanName, "BeanPostProcessor before instantiation of bean failed", ex); } Object beanInstance = doCreateBean(beanName, mbdToUse, args); if (logger.isDebugEnabled()) { logger.debug("Finished creating instance of bean '" + beanName + "'"); } return beanInstance; } // spring // ApplicationContextAwareProcessor.invokeAwareInterfaces,會進行類的默認方法初始化,此處會加載ServiceBean.setApplication初始化方法 private void invokeAwareInterfaces(Object bean) { if (bean instanceof Aware) { if (bean instanceof EnvironmentAware) { ((EnvironmentAware) bean).setEnvironment(this.applicationContext.getEnvironment()); } if (bean instanceof EmbeddedValueResolverAware) { ((EmbeddedValueResolverAware) bean).setEmbeddedValueResolver(this.embeddedValueResolver); } if (bean instanceof ResourceLoaderAware) { ((ResourceLoaderAware) bean).setResourceLoader(this.applicationContext); } if (bean instanceof ApplicationEventPublisherAware) { ((ApplicationEventPublisherAware) bean).setApplicationEventPublisher(this.applicationContext); } if (bean instanceof MessageSourceAware) { ((MessageSourceAware) bean).setMessageSource(this.applicationContext); } if (bean instanceof ApplicationContextAware) { ((ApplicationContextAware) bean).setApplicationContext(this.applicationContext); } } } // 初始化類時,會調用ServiceBean.setApplication @Override public void setApplicationContext(ApplicationContext applicationContext) { this.applicationContext = applicationContext; SpringExtensionFactory.addApplicationContext(applicationContext); if (applicationContext != null) { SPRING_CONTEXT = applicationContext; try { Method method = applicationContext.getClass().getMethod("addApplicationListener", new Class<?>[]{ApplicationListener.class}); // backward compatibility to spring 2.0.1 method.invoke(applicationContext, new Object[]{this}); supportedApplicationListener = true; } catch (Throwable t) { if (applicationContext instanceof AbstractApplicationContext) { try { Method method = AbstractApplicationContext.class.getDeclaredMethod("addListener", new Class<?>[]{ApplicationListener.class}); // backward compatibility to spring 2.0.1 if (!method.isAccessible()) { method.setAccessible(true); } method.invoke(applicationContext, new Object[]{this}); supportedApplicationListener = true; } catch (Throwable t2) { } } } } }
// 添加配置屬性,138+

// Root bean: class [com.alibaba.dubbo.config.spring.ServiceBean] // <bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/> 解析該句里加載dubbo服務 protected static void appendProperties(AbstractConfig config) { if (config == null) { return; } String prefix = "dubbo." + getTagName(config.getClass()) + "."; Method[] methods = config.getClass().getMethods(); for (Method method : methods) { try { String name = method.getName(); if (name.length() > 3 && name.startsWith("set") && Modifier.isPublic(method.getModifiers()) && method.getParameterTypes().length == 1 && isPrimitive(method.getParameterTypes()[0])) { String property = StringUtils.camelToSplitName(name.substring(3, 4).toLowerCase() + name.substring(4), "."); String value = null; if (config.getId() != null && config.getId().length() > 0) { String pn = prefix + config.getId() + "." + property; value = System.getProperty(pn); if (!StringUtils.isBlank(value)) { logger.info("Use System Property " + pn + " to config dubbo"); } } if (value == null || value.length() == 0) { String pn = prefix + property; value = System.getProperty(pn); if (!StringUtils.isBlank(value)) { logger.info("Use System Property " + pn + " to config dubbo"); } } if (value == null || value.length() == 0) { Method getter; try { getter = config.getClass().getMethod("get" + name.substring(3)); } catch (NoSuchMethodException e) { try { getter = config.getClass().getMethod("is" + name.substring(3)); } catch (NoSuchMethodException e2) { getter = null; } } if (getter != null) { if (getter.invoke(config) == null) { if (config.getId() != null && config.getId().length() > 0) { value = ConfigUtils.getProperty(prefix + config.getId() + "." + property); } if (value == null || value.length() == 0) { value = ConfigUtils.getProperty(prefix + property); } if (value == null || value.length() == 0) { String legacyKey = legacyProperties.get(prefix + property); if (legacyKey != null && legacyKey.length() > 0) { value = convertLegacyValue(legacyKey, ConfigUtils.getProperty(legacyKey)); } } } } } if (value != null && value.length() > 0) { method.invoke(config, convertPrimitive(method.getParameterTypes()[0], value)); } } } catch (Exception e) { logger.error(e.getMessage(), e); } } } // AbstractAutowireCapableBeanFactory.invokeInitMethods() @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) { Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false); if ((protocolConfigMap == null || protocolConfigMap.size() == 0) && providerConfigMap.size() > 1) { // backward compatibility List<ProviderConfig> providerConfigs = new ArrayList<ProviderConfig>(); for (ProviderConfig config : providerConfigMap.values()) { if (config.isDefault() != null && config.isDefault().booleanValue()) { providerConfigs.add(config); } } if (!providerConfigs.isEmpty()) { setProviders(providerConfigs); } } else { 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(); } }
// publishEvent(Object event, ResolvableType eventType) protected void publishEvent(Object event, ResolvableType eventType) { Assert.notNull(event, "Event must not be null"); if (logger.isTraceEnabled()) { logger.trace("Publishing event in " + getDisplayName() + ": " + event); } // Decorate event as an ApplicationEvent if necessary ApplicationEvent applicationEvent; if (event instanceof ApplicationEvent) { applicationEvent = (ApplicationEvent) event; } else { applicationEvent = new PayloadApplicationEvent<Object>(this, event); if (eventType == null) { eventType = ((PayloadApplicationEvent) applicationEvent).getResolvableType(); } } // Multicast right now if possible - or lazily once the multicaster is initialized if (this.earlyApplicationEvents != null) { this.earlyApplicationEvents.add(applicationEvent); } else { // 就這一句,開始調用dubbo-bean getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType); } // Publish event via parent context as well... if (this.parent != null) { if (this.parent instanceof AbstractApplicationContext) { ((AbstractApplicationContext) this.parent).publishEvent(event, eventType); } else { this.parent.publishEvent(event); } } } // publishEvent 后, 由 ServiceBean.onApplicationEvent() 接收請求, 進行接口暴露 @Override public void onApplicationEvent(ContextRefreshedEvent event) { if (isDelay() && !isExported() && !isUnexported()) { if (logger.isInfoEnabled()) { logger.info("The service ready on spring started. service: " + getInterface()); } export(); } } @Override public void multicastEvent(final ApplicationEvent event, ResolvableType eventType) { ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event)); for (final ApplicationListener<?> listener : getApplicationListeners(event, type)) { Executor executor = getTaskExecutor(); if (executor != null) { executor.execute(new Runnable() { @Override public void run() { invokeListener(listener, event); } }); } else { invokeListener(listener, event); } } } // one more... public synchronized void export() { if (provider != null) { if (export == null) { export = provider.getExport(); } if (delay == null) { delay = provider.getDelay(); } } if (export != null && !export) { return; } if (delay != null && delay > 0) { delayExportExecutor.schedule(new Runnable() { @Override public void run() { doExport(); } }, delay, TimeUnit.MILLISECONDS); } else { doExport(); } } // one more... protected synchronized void doExport() { if (unexported) { throw new IllegalStateException("Already unexported!"); } if (exported) { return; } exported = true; if (interfaceName == null || interfaceName.length() == 0) { throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!"); } checkDefault(); if (provider != null) { if (application == null) { application = provider.getApplication(); } if (module == null) { module = provider.getModule(); } if (registries == null) { registries = provider.getRegistries(); } if (monitor == null) { monitor = provider.getMonitor(); } if (protocols == null) { protocols = provider.getProtocols(); } } if (module != null) { if (registries == null) { registries = module.getRegistries(); } if (monitor == null) { monitor = module.getMonitor(); } } if (application != null) { if (registries == null) { registries = application.getRegistries(); } if (monitor == null) { monitor = application.getMonitor(); } } if (ref instanceof GenericService) { interfaceClass = GenericService.class; if (StringUtils.isEmpty(generic)) { generic = Boolean.TRUE.toString(); } } else { try { interfaceClass = Class.forName(interfaceName, true, Thread.currentThread() .getContextClassLoader()); } catch (ClassNotFoundException e) { throw new IllegalStateException(e.getMessage(), e); } checkInterfaceAndMethods(interfaceClass, methods); checkRef(); generic = Boolean.FALSE.toString(); } if (local != null) { if ("true".equals(local)) { local = interfaceName + "Local"; } Class<?> localClass; try { localClass = ClassHelper.forNameWithThreadContextClassLoader(local); } catch (ClassNotFoundException e) { throw new IllegalStateException(e.getMessage(), e); } if (!interfaceClass.isAssignableFrom(localClass)) { throw new IllegalStateException("The local implementation class " + localClass.getName() + " not implement interface " + interfaceName); } } if (stub != null) { if ("true".equals(stub)) { stub = interfaceName + "Stub"; } Class<?> stubClass; try { stubClass = ClassHelper.forNameWithThreadContextClassLoader(stub); } catch (ClassNotFoundException e) { throw new IllegalStateException(e.getMessage(), e); } if (!interfaceClass.isAssignableFrom(stubClass)) { throw new IllegalStateException("The stub implementation class " + stubClass.getName() + " not implement interface " + interfaceName); } } // 檢查應用,未加載的配置,屬於重新加載 checkApplication(); // 檢查注冊中心並加載 checkRegistry(); // 檢測注冊使用的協議 checkProtocol(); // 配置當前service appendProperties(this); // 檢查mock返回設置 checkStubAndMock(interfaceClass); if (path == null || path.length() == 0) { path = interfaceName; } // 導出url doExportUrls(); // 初始化本地提供服務的列表 ProviderModel providerModel = new ProviderModel(getUniqueServiceName(), this, ref); ApplicationModel.initProviderModel(getUniqueServiceName(), providerModel); } // 加載配置,如: <dubbo:protocol name="dubbo" port="20880" id="dubbo" /> protected static void appendProperties(AbstractConfig config) { if (config == null) { return; } String prefix = "dubbo." + getTagName(config.getClass()) + "."; Method[] methods = config.getClass().getMethods(); for (Method method : methods) { try { String name = method.getName(); if (name.length() > 3 && name.startsWith("set") && Modifier.isPublic(method.getModifiers()) && method.getParameterTypes().length == 1 && isPrimitive(method.getParameterTypes()[0])) { String property = StringUtils.camelToSplitName(name.substring(3, 4).toLowerCase() + name.substring(4), "."); String value = null; if (config.getId() != null && config.getId().length() > 0) { String pn = prefix + config.getId() + "." + property; value = System.getProperty(pn); if (!StringUtils.isBlank(value)) { logger.info("Use System Property " + pn + " to config dubbo"); } } if (value == null || value.length() == 0) { String pn = prefix + property; value = System.getProperty(pn); if (!StringUtils.isBlank(value)) { logger.info("Use System Property " + pn + " to config dubbo"); } } if (value == null || value.length() == 0) { Method getter; try { getter = config.getClass().getMethod("get" + name.substring(3)); } catch (NoSuchMethodException e) { try { getter = config.getClass().getMethod("is" + name.substring(3)); } catch (NoSuchMethodException e2) { getter = null; } } if (getter != null) { if (getter.invoke(config) == null) { if (config.getId() != null && config.getId().length() > 0) { value = ConfigUtils.getProperty(prefix + config.getId() + "." + property); } if (value == null || value.length() == 0) { value = ConfigUtils.getProperty(prefix + property); } if (value == null || value.length() == 0) { String legacyKey = legacyProperties.get(prefix + property); if (legacyKey != null && legacyKey.length() > 0) { value = convertLegacyValue(legacyKey, ConfigUtils.getProperty(legacyKey)); } } } } } if (value != null && value.length() > 0) { method.invoke(config, convertPrimitive(method.getParameterTypes()[0], value)); } } } catch (Exception e) { logger.error(e.getMessage(), e); } } }
// 展出服務url <dubbo:registry address="multicast://224.5.6.7:1234" id="com.alibaba.dubbo.config.RegistryConfig" />
// 展出服務url <dubbo:registry address="multicast://224.5.6.7:1234" id="com.alibaba.dubbo.config.RegistryConfig" /> @SuppressWarnings({"unchecked", "rawtypes"}) private void doExportUrls() { List<URL> registryURLs = loadRegistries(true); for (ProtocolConfig protocolConfig : protocols) { doExportUrlsFor1Protocol(protocolConfig, registryURLs); } } // 注冊服務 protected List<URL> loadRegistries(boolean provider) { checkRegistry(); List<URL> registryList = new ArrayList<URL>(); if (registries != null && !registries.isEmpty()) { for (RegistryConfig config : registries) { String address = config.getAddress(); if (address == null || address.length() == 0) { address = Constants.ANYHOST_VALUE; } String sysaddress = System.getProperty("dubbo.registry.address"); if (sysaddress != null && sysaddress.length() > 0) { address = sysaddress; } if (address.length() > 0 && !RegistryConfig.NO_AVAILABLE.equalsIgnoreCase(address)) { Map<String, String> map = new HashMap<String, String>(); appendParameters(map, application); appendParameters(map, config); map.put("path", RegistryService.class.getName()); map.put("dubbo", Version.getProtocolVersion()); map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis())); if (ConfigUtils.getPid() > 0) { map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid())); } if (!map.containsKey("protocol")) { if (ExtensionLoader.getExtensionLoader(RegistryFactory.class).hasExtension("remote")) { map.put("protocol", "remote"); } else { map.put("protocol", "dubbo"); } } List<URL> urls = UrlUtils.parseURLs(address, map); for (URL url : urls) { url = url.addParameter(Constants.REGISTRY_KEY, url.getProtocol()); url = url.setProtocol(Constants.REGISTRY_PROTOCOL); if ((provider && url.getParameter(Constants.REGISTER_KEY, true)) || (!provider && url.getParameter(Constants.SUBSCRIBE_KEY, true))) { registryList.add(url); } } } } } return registryList; } // 廣播地址: multicast://224.5.6.7:1234/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.2&pid=14228&qos.port=22222×tamp=1535020152415 // 注冊地址: registry://224.5.6.7:1234/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.2&pid=14228&qos.port=22222®istry=multicast×tamp=1535020152415 // doExportUrlsFor1Protocol(protocolConfig, registryURLs); 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.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE); map.put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion()); map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis())); if (ConfigUtils.getPid() > 0) { map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid())); } appendParameters(map, application); appendParameters(map, module); appendParameters(map, provider, Constants.DEFAULT_KEY); appendParameters(map, protocolConfig); appendParameters(map, this); if (methods != null && !methods.isEmpty()) { for (MethodConfig method : methods) { appendParameters(map, method, method.getName()); String retryKey = method.getName() + ".retry"; if (map.containsKey(retryKey)) { String retryValue = map.remove(retryKey); if ("false".equals(retryValue)) { map.put(method.getName() + ".retries", "0"); } } List<ArgumentConfig> arguments = method.getArguments(); if (arguments != null && !arguments.isEmpty()) { for (ArgumentConfig argument : arguments) { // convert argument type if (argument.getType() != null && argument.getType().length() > 0) { Method[] methods = interfaceClass.getMethods(); // visit all methods if (methods != null && methods.length > 0) { for (int i = 0; i < methods.length; i++) { String methodName = methods[i].getName(); // target the method, and get its signature if (methodName.equals(method.getName())) { Class<?>[] argtypes = methods[i].getParameterTypes(); // one callback in the method if (argument.getIndex() != -1) { if (argtypes[argument.getIndex()].getName().equals(argument.getType())) { appendParameters(map, argument, method.getName() + "." + argument.getIndex()); } else { throw new IllegalArgumentException("argument config error : the index attribute and type attribute not match :index :" + argument.getIndex() + ", type:" + argument.getType()); } } else { // multiple callbacks in the method for (int j = 0; j < argtypes.length; j++) { Class<?> argclazz = argtypes[j]; if (argclazz.getName().equals(argument.getType())) { appendParameters(map, argument, method.getName() + "." + j); if (argument.getIndex() != -1 && argument.getIndex() != j) { throw new IllegalArgumentException("argument config error : the index attribute and type attribute not match :index :" + argument.getIndex() + ", type:" + argument.getType()); } } } } } } } } else if (argument.getIndex() != -1) { appendParameters(map, argument, method.getName() + "." + argument.getIndex()); } else { throw new IllegalArgumentException("argument config must set index or type attribute.eg: <dubbo:argument index='0' .../> or <dubbo:argument type=xxx .../>"); } } } } // end of methods for } if (ProtocolUtils.isGeneric(generic)) { map.put(Constants.GENERIC_KEY, generic); map.put(Constants.METHODS_KEY, Constants.ANY_VALUE); } else { String revision = Version.getVersion(interfaceClass, version); if (revision != null && revision.length() > 0) { map.put("revision", revision); } String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames(); if (methods.length == 0) { logger.warn("NO method found in service interface " + interfaceClass.getName()); map.put(Constants.METHODS_KEY, Constants.ANY_VALUE); } else { map.put(Constants.METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ",")); } } if (!ConfigUtils.isEmpty(token)) { if (ConfigUtils.isDefault(token)) { map.put(Constants.TOKEN_KEY, UUID.randomUUID().toString()); } else { map.put(Constants.TOKEN_KEY, token); } } if (Constants.LOCAL_PROTOCOL.equals(protocolConfig.getName())) { protocolConfig.setRegister(false); map.put("notify", "false"); } // export service String contextPath = protocolConfig.getContextpath(); if ((contextPath == null || contextPath.length() == 0) && provider != null) { contextPath = provider.getContextpath(); } String host = this.findConfigedHosts(protocolConfig, registryURLs, map); Integer port = this.findConfigedPorts(protocolConfig, name, map); URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map); if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class) .hasExtension(url.getProtocol())) { url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class) .getExtension(url.getProtocol()).getConfigurator(url).configure(url); } String scope = url.getParameter(Constants.SCOPE_KEY); // don't export when none is configured if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) { // export to local if the config is not remote (export to remote only when config is remote) if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) { // 本地運行時會注冊服務到本地 exportLocal(url); } // export to remote if the config is not local (export to local only when config is local) if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) { if (logger.isInfoEnabled()) { logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url); } if (registryURLs != null && !registryURLs.isEmpty()) { for (URL registryURL : registryURLs) { url = url.addParameterIfAbsent(Constants.DYNAMIC_KEY, registryURL.getParameter(Constants.DYNAMIC_KEY)); URL monitorUrl = loadMonitor(registryURL); if (monitorUrl != null) { url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString()); } if (logger.isInfoEnabled()) { logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL); } // For providers, this is used to enable custom proxy to generate invoker String proxy = url.getParameter(Constants.PROXY_KEY); if (StringUtils.isNotEmpty(proxy)) { registryURL = registryURL.addParameter(Constants.PROXY_KEY, proxy); } Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); // subscribeUrl: provider://192.168.11.6:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&category=configurators&check=false&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=14228&side=provider×tamp=1535020596190 // registerUrl: dubbo://192.168.11.6:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=14228&side=provider×tamp=1535020596190 Exporter<?> exporter = protocol.export(wrapperInvoker); exporters.add(exporter); } } else { Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url); DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); Exporter<?> exporter = protocol.export(wrapperInvoker); exporters.add(exporter); } } } this.urls.add(url); } // 導出服務到本地 @SuppressWarnings({"unchecked", "rawtypes"}) private void exportLocal(URL url) { if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) { URL local = URL.valueOf(url.toFullString()) .setProtocol(Constants.LOCAL_PROTOCOL) .setHost(LOCALHOST) .setPort(0); ServiceClassHolder.getInstance().pushServiceClass(getServiceClass(ref)); // 執行真正的導出服務暴露 // 鏈接如下: injvm://127.0.0.1/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=192.168.11.6&bind.port=20880&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=14228&qos.port=22222&side=provider×tamp=1535020596190 Exporter<?> exporter = protocol.export( proxyFactory.getInvoker(ref, (Class) interfaceClass, local)); exporters.add(exporter); logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry"); } }
// 以上是通過本方廣播的方式進行提供dubbo服務,使其運行在本地jvm中的運行調用鏈。
// 下面來看看dubbo以遠程調用形式進行提供服務的情況
// DubboProtocol.export(), 以dubbo協議遠程暴露dubbo @Override public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { URL url = invoker.getUrl(); // export service. String key = serviceKey(url); DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap); exporterMap.put(key, exporter); //export an stub service for dispatching event Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT); Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false); if (isStubSupportEvent && !isCallbackservice) { String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY); if (stubServiceMethods == null || stubServiceMethods.length() == 0) { if (logger.isWarnEnabled()) { logger.warn(new IllegalStateException("consumer [" + url.getParameter(Constants.INTERFACE_KEY) + "], has set stubproxy support event ,but no stub methods founded.")); } } else { stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods); } } openServer(url); optimizeSerialization(url); return exporter; } private void openServer(URL url) { // find server. String key = url.getAddress(); //client can export a service which's only for server to invoke boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true); if (isServer) { ExchangeServer server = serverMap.get(key); if (server == null) { serverMap.put(key, createServer(url)); } else { // server supports reset, use together with override server.reset(url); } } } private ExchangeServer createServer(URL url) { // send readonly event when server closes, it's enabled by default url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()); // enable heartbeat by default url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT)); String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER); if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) throw new RpcException("Unsupported server type: " + str + ", url: " + url); url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME); ExchangeServer server; try { server = Exchangers.bind(url, requestHandler); } catch (RemotingException e) { throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); } str = url.getParameter(Constants.CLIENT_KEY); if (str != null && str.length() > 0) { Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(); if (!supportedTypes.contains(str)) { throw new RpcException("Unsupported client type: " + str); } } return server; } // 遠程服務框架中dubbo提供的有幾種, netty/netty4/Grizzly, 我們就看一下使用netty的方式說明下 public NettyServer(URL url, ChannelHandler handler) throws RemotingException { super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME))); } @Override protected void doOpen() throws Throwable { NettyHelper.setNettyLoggerFactory(); ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true)); ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true)); ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS)); bootstrap = new ServerBootstrap(channelFactory); final NettyHandler nettyHandler = new NettyHandler(getUrl(), this); channels = nettyHandler.getChannels(); // https://issues.jboss.org/browse/NETTY-365 // https://issues.jboss.org/browse/NETTY-379 // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true)); bootstrap.setOption("child.tcpNoDelay", true); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() { NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); ChannelPipeline pipeline = Channels.pipeline(); /*int idleTimeout = getIdleTimeout(); if (idleTimeout > 10000) { pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0)); }*/ pipeline.addLast("decoder", adapter.getDecoder()); pipeline.addLast("encoder", adapter.getEncoder()); pipeline.addLast("handler", nettyHandler); return pipeline; } }); // bind channel = bootstrap.bind(getBindAddress()); } // server 創建后,對上層是不可見的,上層只知道 Transporter/ExchangeServer @Override public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); } // HeaderExchangeServer設置好server后,啟動一個心跳定時器與注冊中心保持關聯 public HeaderExchangeServer(Server server) { if (server == null) { throw new IllegalArgumentException("server == null"); } this.server = server; this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0); this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3); if (heartbeatTimeout < heartbeat * 2) { throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2"); } startHeartbeatTimer(); }
總結下代碼加載過程:
到此,完整的server的啟動就完成了,就等待客戶端的連接了。