Dubbo初始化过程分析

黄克明

Dubbo是阿里巴巴开源的基于java实现的高性能、透明化的RPC框架。深入了解dubbo源码,有助于快速定位问题、高效实现自定义拓展。本文以dubbo服务端初始化过程为例,分析dubbo怎么从配置转化成可被调用的服务。

以典型的服务端结合spring配置为例:

<!-- 提供方应用信息,用于计算依赖关系 -->  
<dubbo:application name="demo-provider"/>  
<!-- 用dubbo协议在20880端口暴露服务 -->  
<dubbo:protocol name="dubbo" port="20880"/>  
<!-- 使用zookeeper注册中心暴露服务地址 -->  
<dubbo:registry address="zookeeper://127.0.0.1:1234" id="registry"/>  
<!-- 默认的服务端配置 -->  
<dubbo:provider registry="registry" retries="0" timeout="5000"/>  
<!-- 和本地bean一样实现服务 -->  
<bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/>  
<!-- 声明需要暴露的服务接口 -->  
<dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService"/>  

在dubbo命名空间下定义了一系列xml节点,如:application、protocol、registry、provider、service等,dubbo通过实现spring提供的NamespaceHandler接口,向spring注册BeanDefinition解析器,使spring能识别dubbo命名空间下的节点;通过实现BeanDefinitionParser接口,使spring能解析各节点的具体配置。

DubboNamespaceHandler.java

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 DubboBeanDefinitionParser(AnnotationBean.class, true));
}

由上代码可以看出,各个节点最终被转化为各种Bean,配置的各种属性也被转化为Bean的属性。从Bean的类型可以看出,大部分Bean只用于提供dubbo的运行参数,本文重点分析ServiceBean

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware {  
    ...
    public void afterPropertiesSet() {}
    ...
    public void onApplicationEvent(ApplicationEvent event) {}
    ...
    public void destroy() {}
}

ServiceBean实现了spring的InitializingBeanDisposableBeanApplicationListener等接口,实现了afterPropertiesSet()destroy()onApplicationEvent()等典型方法,这里便是dubbo和spring整合的关键,一般第三方框架基本都是通过这几个接口和spring整合的。

public void afterPropertiesSet() throws Exception {  
    ...
    setProvider(providerConfig);
    ...
    setApplication(applicationConfig);
    ...
    setModule(moduleConfig);
    ...
    setRegistries(registryConfigs);
    ...
    setMonitor(monitorConfig);
    ...
    setProtocols(protocolConfigs);
    ...
    if (!isDelay()) {
        export();
    }
    ...
}

afterPropertiesSet()主要用来注入各种configBean,便于服务注册过程中各种参数的获取,注意看最后几行代码,大意是如果不延迟,就立即注册和暴露服务。

查看isDelay()方法:

private boolean isDelay() {  
    Integer delay = getDelay();
    ProviderConfig provider = getProvider();
    if (delay == null && provider != null) {
        delay = provider.getDelay();
    }
    return supportedApplicationListener && (delay == null || delay.intValue() == -1);
}

先从ServiceConfig获取delay属性,如果为null则获取ProviderConfigdelay属性,最后如果还是null或配置为-1表示延迟暴露服务。可见dubbo获取运行参数的层级,便于更精确化的配置各种参数。

通过supportedApplicationListener可以猜到服务延迟暴露是通过spring容器的监听器触发的。个人更倾向于明确设置delay=-1或者所有层级都不配置,因为如果提早暴露服务,此时其他的spring bean可能还未初始化完成,而暴露出去的服务大部分情况下依赖于spring的其他bean来实现业务功能,如果接收到客户端的请求,难免会出现各种异常。

public void onApplicationEvent(ApplicationEvent event) {  
    if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
        if (isDelay() && !isExported() && !isUnexported()) {
            if (logger.isInfoEnabled()) {
                ...
            }
            export();
        }
    }
}

通过重重检查,最后调用了export()方法,沿着方法栈查看,各种参数检查和设置,最终调用doExportUrls()

private void doExportUrls() {  
    List<URL> registryURLs = loadRegistries(true);
    for (ProtocolConfig protocolConfig : protocols) {
        doExportUrlsFor1Protocol(protocolConfig, registryURLs);
    }
}

可以看出dubbo同一个服务支持多种服务协议、支持向多种注册中心注册,很方便同一功能由各种不同实现方式的客户端调用。

ServiceConfig.java

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {  
    //整合各种url所需参数
    map.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE);
    map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
    map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
    //方法级别细粒度配置
    if (methods != null && methods.size() > 0) {
        for (MethodConfig method : methods) {
            appendParameters(map, method, method.getName());
        }
    }
    //服务提供的方法
    String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
        if (methods.length == 0) {
            logger.warn("NO method found in service interface " + interfaceClass.getName());
            map.put("methods", Constants.ANY_VALUE);
        } else {
            map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
        }
    //服务URL
    URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);

    //根据scope配置,判断暴露远程服务还是本地服务
    String scope = url.getParameter(Constants.SCOPE_KEY);
    //注册为本地服务
    if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
            exportLocal(url);
    }
    //注册为远程服务
    for (URL registryURL : registryURLs) {
        url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
        URL monitorUrl = loadMonitor(registryURL);
        if (monitorUrl != null) {
            url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
        }
        //⑴使用ProxyFactory将服务实现封装成一个Invoker对象
        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
        //⑵根据指定协议本地暴露和向注册中心注册服务
        Exporter<?> exporter = protocol.export(invoker);
        //用于unexport
        exporters.add(exporter);
     }
    this.urls.add(url);
}

上面这个方法虽然很长,但是不是很复杂,无非就是整合url所需参数,然后生成服务端代理,向注册中心注册此url和暴露服务。

下面重点分析⑴和⑵处,从这里开始深入dubbo内部实现。

这两处调用有类似之处,都是通过扩展机制根据不同参数配置得到不同的具体实现对象,扩展解析过程这里不做详细分析。

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

getAdaptiveExtension()会通过拼接源码的方式动态生成Class,生成的Class方法中会获取url中的参数来构建合适的具体实现对象,如果url中未配置,则使用@SPI配置的默认值。

getAdaptiveExtension()调用栈:

ExtensionLoader<T>.getAdaptiveExtension()  
    ExtensionLoader<T>.createAdaptiveExtension()
        ExtensionLoader<T>.getAdaptiveExtensionClass()
            ExtensionLoader<T>.createAdaptiveExtensionClass()
                ExtensionLoader<T>.createAdaptiveExtensionClassCode()

查看ProxyFactoryProtocol接口,默认ProxyFactory实现为JavassistProxyFactory,默认Protocol实现为DubboProtocol

//默认javassist
@SPI("javassist")
public interface ProxyFactory {  
    ...
}

//默认dubbo
@SPI("dubbo")
public interface Protocol {  
    ...
}
代码⑴处分析

未配置proxy实现方式,所以使用JavassistProxyFactory

JavassistProxyFactory.getInvoker()通过拼接源码,编译实例化后生成Wrapper的实现类,WrapperinvokeMethod()方法就是调用interface的具体实现方法(业务方法)。通过代理的方式,就可以在调用实际方法前后增加各种逻辑,也就是AOP思想。

JavassistProxyFactory.java

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {  
    // TODO Wrapper类不能正确处理带$的类名
    final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
    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);
        }
    };
}

查看动态生成wrapper对象的源码:

public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws java.lang.reflect.InvocationTargetException {  
    com.alibaba.dubbo.demo.provider.DemoServiceImpl w;
    try {
        w = ((com.alibaba.dubbo.demo.provider.DemoServiceImpl) $1);
    } catch (Throwable e) {
        throw new IllegalArgumentException(e);
    }
    try {
        if ("sayHello".equals($2) && $3.length == 1) {
            return ($w) w.sayHello((java.lang.String) $4[0]);
        }
    } catch (Throwable e) {
        throw new java.lang.reflect.InvocationTargetException(e);
    }
    throw new com.alibaba.dubbo.common.bytecode.NoSuchMethodException("Not found method \"" + $2 + "\" in class com.alibaba.dubbo.demo.provider.DemoServiceImpl.");
}

可以看到w.sayHello()这就是直接通过服务的实现对象调用具体方法,并不是通过反射,效率会高些。默认使用Javassist而不是JDK动态代理也是出于效率的考虑。

代码⑵处分析

把得到的invoker传入Protocolexport()方法,猜想服务端接收到请求后,会调用invoker.invoke(),执行业务逻辑。

这里我们有必要看下针对Protocol.classgetAdaptiveExtension()最终调用createAdaptiveExtensionClassCode()生成的源码:

    public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException {
        if (arg0 == null)
            throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
        if (arg0.getUrl() == null)
            throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");
        com.alibaba.dubbo.common.URL url = arg0.getUrl();
        String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
        com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
        return extension.export(arg0);
    }

注意看变量extName的那行,是根据url的协议类型获得对应的Protocol,服务注册过程中传入的url是registry://开头的url,可以称为注册url,发布url整体作为参数拼接在注册url的export参数。

所以此处应该实例化的是RegistryProtocol的对象,但是实际上此处获取到的是RegistryProtocol的装饰对象ProtocolListenerWrapper,这又是从哪儿来的呢?

深入getExtension()方法,在调用createExtension()时,如果有wrapperClass,则把原始对象作为构造器参数构建装饰对象。

Set<Class<?>> wrapperClasses = cachedWrapperClasses;  
if (wrapperClasses != null && wrapperClasses.size() > 0) {  
    for (Class<?> wrapperClass : wrapperClasses) {
        instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance));
    }
}

cachedWrapperClasses又从哪儿来的呢?在加载扩展配置文件时,也就是loadFile()方法有一段:

try {  
    clazz.getConstructor(type);
    Set<Class<?>> wrappers = cachedWrapperClasses;
    if (wrappers == null) {
        cachedWrapperClasses = new ConcurrentHashSet<Class<?>>();
        wrappers = cachedWrapperClasses;
    }
    wrappers.add(clazz);
} catch (NoSuchMethodException e) {
    ...
}

在解析扩展配置文件(/META-INF/dubbo/*)过程中,把配置加载成Class时,会执行上面代码,会检查是否存在以原始Protocol类型为参数类型的构造方法,如果不抛出NoSuchMethodException,说明此Class为原始类型的装饰类型,加入到cachedWrapperClasses中。

/META-INF/dubbo/internal/com.alibaba.dubbo.rpc.Protocol中配置有:

filter=com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper  
listener=com.alibaba.dubbo.rpc.protocol.ProtocolListenerWrapper  

RegistryProtocol会被ProtocolFilterWrapperProtocolListenerWrapper装饰,分别用来实现拦截器和监听器功能,查看这两个Wrapper的代码可以看出,对于注册url都做了特别处理,向注册中心发布url不会触发拦截器和监听器功能,只有在真正暴露服务时才会注册拦截器,触发监听器。

ProtocolFilterWrapper.java

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {  
    if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
        return protocol.export(invoker);
    }
    return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
}

最终调到RegistryProtocolexport()代码:

RegistryProtocol.java

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {  
    //暴露服务的监听
    final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
    ...
    //注册服务
    final Registry registry = getRegistry(originInvoker);
    registry.register(registedProviderUrl);
    ...
    // 订阅注册中心服务
    registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
    ...
    };
}

可以看到,先开启本地服务监听,然后把发布url发布到注册中心,如zookeeper。发布到注册中心就先分析到这一步,可以猜测也是根据url的参数,选择哪种注册中心发布。

下面看看doLocalExport()方法:

final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));  
exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);  

这里的protocol也是getAdaptiveExtension()生成的对象Protocol$Adpative,是在RegistryProtocol对象创建过程中注入进来的,可以跟踪createExtension()injectExtension()方法。

调用protocol.export()又回到上面那段动态生成的源码逻辑,这次得到的ProtocolDubboProtocol,又被ProtocolFilterWrapperProtocolListenerWrapper装饰,在export的过程中会形成Filter链和触发Listener执行。

DubboProtocol.java

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);
    ...
    //建立服务监听
    openServer(url);
    return exporter;
}

可以看到,终于要开启本地端口监听暴露服务了。

总结

本文就先暂时分析到此处,本文大致分析了dubbo初始化过程,涉及dubbo十层结构中的config(配置层)、proxy(服务代理层)、registry(注册中心层)、protocol(远程调用层),还有其他一些结构需要读者自行分析,本文仅起到抛砖引玉的作用。