【DUBBO】服务暴露

2018-11-28 本文已影响33人 spilledyear

在Spring将xml中的各个标签解析成一个个BeanDefinition之后，接下来就是根据BeanDefinition实例化Bean。有关于dubbo标签中对应的一些Bean，是用于维护一些配置信息，比如：ApplicationConfig、ProtocolConfig、ProviderConfig等，这些Bean的实例化细节就不过多介绍，重点关注ServiceBean和ReferenceBean，ServiceBean也就是对应xml中的<dubbo:service/>，ReferenceBean对应着xml中的<dubbo:reference/>。

本文主要是介绍ServiceBean的实例化过程，主要涉及到服务暴露，有关于ReferenceBean的相关内容，将在下一篇文章介绍。

ServiceBean实例化

ServiceBean实现的接口比较多，如下：

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent>, BeanNameAware {

    private static final long serialVersionUID = 213195494150089726L;

    private static transient ApplicationContext SPRING_CONTEXT;

    private final transient Service service;

    private transient ApplicationContext applicationContext;

    private transient String beanName;

    private transient boolean supportedApplicationListener;

    public ServiceBean() {
        super();
        this.service = null;
    }

    public ServiceBean(Service service) {
        super(service);
        this.service = service;
    }
}

都是Spring中的一些扩展接口，通过这种方式，可以在初始化Bean前后插入一些逻辑，一个个分析：

实现DisposableBean

实现DisposableBean接口或者在方法上添加了@PreDestroy注解，在销毁bean之前可以进行一些操作，这里是实现了DisposableBean方法，所以在销毁Bean之前会执行destroy方法，这里是空实现：

@Override
public void destroy() throws Exception {
}

实现BeanNameAware

实现了BeanNameAware接口，在初始化Bean之前会调用了setBeanName方法，即设置beanName：

@Override
public void setBeanName(String name) {
    this.beanName = name;
}

实现ApplicationContextAware

实现了接口ApplicationContextAware接口，所以会在当前Bean初始化之前调用setApplicationContext方法，所以将ApplicationContext注入到当前Bean。

public void setApplicationContext(ApplicationContext applicationContext) {
    this.applicationContext = applicationContext;
    SpringExtensionFactory.addApplicationContext(applicationContext);
    if (applicationContext != null) {
        SPRING_CONTEXT = applicationContext;
        try {
            // 这是什么骚操作啊？为什么要调用applicationContext的addApplicationListener方法，spring在启动的时候不是会调用registerListeners
            // 方法注册所有的ApplicationListener吗？难道仅仅是为了判断是否支持监听，即 supportedApplicationListener？
            Method method = applicationContext.getClass().getMethod("addApplicationListener", ApplicationListener.class); 
            method.invoke(applicationContext, this);
            supportedApplicationListener = true;
        } catch (Throwable t) {
            if (applicationContext instanceof AbstractApplicationContext) {
                try {
                    Method method = AbstractApplicationContext.class.getDeclaredMethod("addListener", ApplicationListener.class); 
                    if (!method.isAccessible()) {
                        method.setAccessible(true);
                    }
                    method.invoke(applicationContext, this);
                    supportedApplicationListener = true;
                } catch (Throwable t2) {
                }
            }
        }
    }
}

实现ApplicationListener

实现了ApplicationListener接口，表明这是一个监听者，其实就是观察者模式。Spring在启动的时候，会注册所有的ApplicationListener，然后在发出ApplicationEvent事件的时候，根据不同的事情类型，执行不同的ApplicationListener的onApplicationEvent方法。这里监听的是ContextRefreshedEvent时间，表明ApplicationContext被初始化或刷新时，该事件被发布。即：ApplicationContext被初始化或刷新时，会执行这里的onApplicationEvent方法：

@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
    if (isDelay() && !isExported() && !isUnexported()) {
        if (logger.isInfoEnabled()) {
            logger.info("The service ready on spring started. service: " + getInterface());
        }
        // 暴露服务， 该方法位于ServiceConfig中
        export();
    }
}

在Spring容器初始化或刷新时的时候，可能会发布服务，这里就不分析，因为在afterPropertiesSet方法中会涉及到这一块内容。

实现InitializingBean

实现InitializingBean了接口，所以在Bean初始化之后会执行afterPropertiesSet方法。通过给方法加@PostConstruct注解或者在xml中制定init-method属性，也可以达到同样的目的。这个方法比较长，这里主要是做一些前置处理，主要是用来注入各种configBean，便于服务注册过程中获取各种参数。这个方法的比较长，所以这里删掉了一些非关键代码，只留下大致流程：

public void afterPropertiesSet() throws Exception {

    // ...... 组装 providerConfig
    setProvider(providerConfig);

    // ...... 组装 applicationConfig
    setApplication(applicationConfig);
    
    // ...... 组装 moduleConfig
    setModule(moduleConfig);

    // ...... 组装 registryConfigs
    super.setRegistries(registryConfigs);

    // ...... 组装 monitorConfig
    setMonitor(monitorConfig);

    // ...... 组装 protocolConfigs
    super.setProtocols(protocolConfigs);

    setPath(beanName);

    // isDelay用于判断是否延迟发布服务，先从ServiceConfig获取delay属性，如果为null则获取ProviderConfig的delay属性，最后如果还是null或配置为-1表示延迟暴露服务。
    if (!isDelay()) {
        // 发布服务
        export();
    }
}

isDelay方法用于判断是否延迟发布服务，先从ServiceConfig获取delay属性，如果为null则获取ProviderConfig的delay属性，最后如果还是null或配置为-1表示延迟暴露服务。下面将重点介绍服务发布

服务暴露

服务发布是一个非常重要的流程，方法调用流程如下：ServiceConfig.export() =》 ServiceConfig.doExport() =》ServiceConfig.doExportUrls() =》 ServiceConfig.doExportUrlsFor1Protocol

private static final ScheduledExecutorService delayExportExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("DubboServiceDelayExporter", true));

public synchronized void export() {
    ......

    if (delay != null && delay > 0) {
        // 延迟发布
        delayExportExecutor.schedule(new Runnable() {
            @Override
            public void run() {
                doExport();
            }
        }, delay, TimeUnit.MILLISECONDS);
    } else {
        doExport();
    }
}

如果需要延迟发布服务，就通过newSingleThreadScheduledExecutor延迟发布该服务，然后调用doExport方法。

doExport方法中主要是各个各样的校验，然后调用了doExportUrls方法，同一个服务支持多种服务协议、支持向多种注册中心注册。

private void doExportUrls() {
    // 获取所有注册中心URL
    List<URL> registryURLs = loadRegistries(true);
    for (ProtocolConfig protocolConfig : protocols) {
        doExportUrlsFor1Protocol(protocolConfig, registryURLs);
    }
}

doExportUrlsFor1Protocol
这个方法非常长，但是很大一部分工作都是构建url参数，为啥写那么长一个方法？这里省略部分代码

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
    String name = protocolConfig.getName();
    if (name == null || name.length() == 0) {
        name = "dubbo";
    }

    // 设置url中的各个参数
    Map<String, String> map = new HashMap<>();
    // side = provider
    map.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE);
    // timestamp
    map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
    // dubbo协议的版本
    map.put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion());
    // pid
    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);


......中间省略的这一部分内容分都是在构建URl参数......


    // 从这里开始暴露服务
    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 url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);

    // todo 这里没看懂，不知道什么意思
    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);
    // 如果scope设置成none，就不暴露服务
    if (!Constants.SCOPE_NONE.equalsIgnoreCase(scope)) {
        // 当scope不是remote的时候，暴露本地服务
        if (!Constants.SCOPE_REMOTE.equalsIgnoreCase(scope)) {
            exportLocal(url);
        }
       // 如果scope != local，暴露远程服务，所以如果没有配置scope，则先暴露本地服务，然后暴露远程服务
        if (!Constants.SCOPE_LOCAL.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);
                    }

                    // private static final ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

                    // private static final Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
                    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);
}

构建好URL之后，会根据scope选择暴露本地服务还是远程服务。

本地暴露

为何要本地暴露？加入服务消费者和服务提供者都在一个JVM进行内，消费者刚好调用了本JVM进程之内的服务提供者，这时候就可以通过反射的机制执行本地调用，这时候就暴露服务的时候就没必要去开一个NIO Server了，提高效率。

private void exportLocal(URL url) {
    if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
        // 将非injvm协议，转换成injvm协议，即本地协议
        URL local = URL.valueOf(url.toFullString())
                .setProtocol(Constants.LOCAL_PROTOCOL)
                .setHost(LOCALHOST)
                .setPort(0);
        // ServiceClassHolder是用来保存当前服务接口实例ref对应的Class的,是一个简单的单例实现
        ServiceClassHolder.getInstance().pushServiceClass(getServiceClass(ref));
        // 这里的protocol 是 InjvmProtocol
        Exporter<?> exporter = protocol.export(
                proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
        exporters.add(exporter);
        logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry");
    }
}

在上面的代码中，先将将非injvm协议，转换成injvm协议，例如

// 转换前
dubbo://10.10.129.71:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.10.129.71&bind.port=20880&dubbo=2.0.2&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=11528&qos.port=22222&side=provider&timestamp=1543753540296

// 转换后
injvm://127.0.0.1/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.10.129.71&bind.port=20880&dubbo=2.0.2&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=11528&qos.port=22222&side=provider&timestamp=1543753540296

然后执行InjvmProtocol的export方法，直接创建一个InjvmExporter对象返回ServiceConfig

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
    return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(), exporterMap);
}

下面进行一个简单的测试
1、提供方和消费方位于不同JVM进程
启动两个应用，分别为provider1和consumer1

<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 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="org.apache.dubbo.demo.provider.DemoServiceImpl"/>
    <!-- 使用注册中心 -->
    <dubbo:registry address="zookeeper://localhost:2181"/>
    <dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService"/>
</beans>

public class Provider {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-provider.xml"});
        context.start();
        System.in.read();
}

public class Provider {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-provider.xml"});
        context.start();
        System.in.read();

         可用于测试本地调用
        DemoService demoService = (DemoService) context.getBean("demoService2");
        while (true) {
            try {
                Thread.sleep(1000);
                String hello = demoService.sayHello("world"); // call remote method
                System.out.println(hello); // get result
            } catch (Throwable throwable) {
                throwable.printStackTrace();
            }
        }
    }
}

2、另开一个服务提供方进程
修改dubbo协议端口，对实现类的输出内容稍作修改，启动provider，称此进程为provider2，观察此时控制台的输出内容。

3、提供方和消费方位于相同JVM进程
启动一个应用，里面包括了提供方和调用方，分别为provider100和consumer100，这时候修改一下配置，同时将服务提供方是实现类里面的输出内容稍作修改，方便对比。

<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 dubbo protocol to export service on port 20880 -->
    <dubbo:protocol name="dubbo" port="20881"/>
    <!-- service implementation, as same as regular local bean -->
    <bean id="demoService" class="org.apache.dubbo.demo.provider.DemoServiceImpl"/>
    <!-- 不使用注册中心，即本地调用 -->
    <dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" registry="N/A" scope="local"/>
    <!-- 消费者，测试方便 -->
    <dubbo:reference id="demoService2" check="false" interface="org.apache.dubbo.demo.DemoService"/>
</beans>

public class Provider {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-provider.xml"});
        context.start();
        System.in.read();
}

4、结果分析
对于consumer1，因为它是在一个独立的JVM运行，所以它在消费服务提供方的时候需要通过网络层，而此时的provider1和provider2因为就相当于建立了一个服务集群，该集群中有两个服务实例。在consumer1消费的时候，通过dubbo内部的负载均衡机制对provider1和provider2进行消费。所以，consumer1中的控制台信息，有时候是输出provider1返回的信息，有时候是返回provider2返回的信息。

而provider100和consumer100因为是在同一个JVM进程内，并且没有注册中心，provider100被设置成本地暴露，consumer100对provider1和provider2是无法感知的，因此consumer100中输出的信息只有provider100返回的信息。

远程暴露

如果注册了多个注册中心，会将服务发布到所有注册中心。在服务远程发布流程中，核心就在以下代码总

// private static final ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

// private static final Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
Exporter<?> exporter = protocol.export(wrapperInvoker);
exporters.add(exporter);

如果对dubbo中的SPI机制很了解的话，就会知道这里和SPI相关，如果不是很清楚，建议去看看我的另一片关于SPI的文章。下面将分别介绍ProxyFactory和Protocol这两个扩展的getAdaptiveExtension方法。

ProxyFactory#getAdaptiveExtension

ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

先简单看看ProxyFactory 扩展，当URL中属性proxy没有的时候，将使用javassist对应的扩展实现类，即 JavassistProxyFactory

@SPI("javassist")
public interface ProxyFactory {
    // Constants.PROXY_KEY 即 proxy

    @Adaptive({Constants.PROXY_KEY})
    <T> T getProxy(Invoker<T> invoker) throws RpcException;

    @Adaptive({Constants.PROXY_KEY})
    <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException;

    @Adaptive({Constants.PROXY_KEY})
    <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException;
}

下面是根据在getInvoker方法上的@Adaptive注解动态生成的代码：

public class ProxyFactory$Adaptive implements org.apache.dubbo.rpc.ProxyFactory {
    private static final org.apache.dubbo.common.logger.Logger logger = org.apache.dubbo.common.logger.LoggerFactory.getLogger(ExtensionLoader.class);
    private java.util.concurrent.atomic.AtomicInteger count = new java.util.concurrent.atomic.AtomicInteger(0);

    public java.lang.Object getProxy(org.apache.dubbo.rpc.Invoker arg0) throws org.apache.dubbo.rpc.RpcException {
        if (arg0 == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null");
        if (arg0.getUrl() == null)
            throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument getUrl() == null");
        org.apache.dubbo.common.URL url = arg0.getUrl();
        String extName = url.getParameter("proxy", "javassist");
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(org.apache.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
        org.apache.dubbo.rpc.ProxyFactory extension = null;
        try {
            extension = (org.apache.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        } catch (Exception e) {
            if (count.incrementAndGet() == 1) {
                logger.warn("Failed to find extension named " + extName + " for type org.apache.dubbo.rpc.ProxyFactory, will use default extension javassist instead.", e);
            }
            extension = (org.apache.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.ProxyFactory.class).getExtension("javassist");
        }
        return extension.getProxy(arg0);
    }

    public java.lang.Object getProxy(org.apache.dubbo.rpc.Invoker arg0, boolean arg1) throws org.apache.dubbo.rpc.RpcException {
        if (arg0 == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null");
        if (arg0.getUrl() == null)
            throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument getUrl() == null");
        org.apache.dubbo.common.URL url = arg0.getUrl();
        String extName = url.getParameter("proxy", "javassist");
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(org.apache.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
        org.apache.dubbo.rpc.ProxyFactory extension = null;
        try {
            extension = (org.apache.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        } catch (Exception e) {
            if (count.incrementAndGet() == 1) {
                logger.warn("Failed to find extension named " + extName + " for type org.apache.dubbo.rpc.ProxyFactory, will use default extension javassist instead.", e);
            }
            extension = (org.apache.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.ProxyFactory.class).getExtension("javassist");
        }
        return extension.getProxy(arg0, arg1);
    }

    public org.apache.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, java.lang.Class arg1, org.apache.dubbo.common.URL arg2) throws org.apache.dubbo.rpc.RpcException {
        if (arg2 == null) throw new IllegalArgumentException("url == null");
        org.apache.dubbo.common.URL url = arg2;
        String extName = url.getParameter("proxy", "javassist");
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(org.apache.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
        org.apache.dubbo.rpc.ProxyFactory extension = null;
        try {
            extension = (org.apache.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        } catch (Exception e) {
            if (count.incrementAndGet() == 1) {
                logger.warn("Failed to find extension named " + extName + " for type org.apache.dubbo.rpc.ProxyFactory, will use default extension javassist instead.", e);
            }
            extension = (org.apache.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.ProxyFactory.class).getExtension("javassist");
        }
        return extension.getInvoker(arg0, arg1, arg2);
    }
}

经过观察，发现ProxyFactory扩展实现中有Wrapper扩展：StubProxyFactoryWrapper，所以在执行以下代码的时候

Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));

实际上的执行顺序是这样的：StubProxyFactoryWrapper =》JavassistProxyFactory

先看看StubProxyFactoryWrapper的getInvoker方法，发现其内部仅仅是简单调用了代理类的getInvoker方法

@Override
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException {
    return proxyFactory.getInvoker(proxy, type, url);
}

再看看JavassistProxyFactory的getInvoker方法

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
    // 根据实现类的名称，动态生成源代码，然后编译创建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);
        }
    };
}


public static Wrapper getWrapper(Class<?> c) {
    // 这里通过ClassGenerator的内部类DC来标识动态class，所以只要是ClassGenerator.DC的子类都属于动态类
    while (ClassGenerator.isDynamicClass(c)) 
        c = c.getSuperclass();

    if (c == Object.class)
        return OBJECT_WRAPPER;

    Wrapper ret = WRAPPER_MAP.get(c);
    if (ret == null) {
        // 动态生成源代码，然后编译创建对象
        ret = makeWrapper(c);
        WRAPPER_MAP.put(c, ret);
    }
    return ret;
}

// 动态生成源代码，然后编译创建对象
private static Wrapper makeWrapper(Class<?> c) {
    // 判断Class是否为原始类型：boolean、char、byte、short、int、long、float、double
    if (c.isPrimitive())
        throw new IllegalArgumentException("Can not create wrapper for primitive type: " + c);

    String name = c.getName();
    ClassLoader cl = ClassHelper.getClassLoader(c);

    StringBuilder c1 = new StringBuilder("public void setPropertyValue(Object o, String n, Object v){ ");
    c1.append(name).append(" w; try{ w = ((").append(name).append(")$1); }catch(Throwable e){ throw new IllegalArgumentException(e); }");

    ......省略很多代码......
    

    // 开始创建Wrapper对象
    long id = WRAPPER_CLASS_COUNTER.getAndIncrement();
    ClassGenerator cc = ClassGenerator.newInstance(cl);
    cc.setClassName((Modifier.isPublic(c.getModifiers()) ? Wrapper.class.getName() : c.getName() + "$sw") + id);
    cc.setSuperClass(Wrapper.class);

    cc.addDefaultConstructor();
    cc.addField("public static String[] pns;"); 
    cc.addMethod("public String[] getPropertyNames(){ return pns; }");

    ......省略很多代码......

    try {
        Class<?> wc = cc.toClass();
        // setup static field.
        wc.getField("pts").set(null, pts);
        wc.getField("pns").set(null, pts.keySet().toArray(new String[0]));
        wc.getField("mns").set(null, mns.toArray(new String[0]));
        wc.getField("dmns").set(null, dmns.toArray(new String[0]));
        int ix = 0;
        for (Method m : ms.values())
            wc.getField("mts" + ix++).set(null, m.getParameterTypes());
        return (Wrapper) wc.newInstance();
    } catch (RuntimeException e) {
        throw e;
    } catch (Throwable e) {
        throw new RuntimeException(e.getMessage(), e);
    } finally {
        cc.release();
        ms.clear();
        mns.clear();
        dmns.clear();
    }
}

然后返回一个AbstractProxyInvoker对象，然后将返回的和ServiceConfig封装在一个DelegateProviderMetaDataInvoker对象中，作为protocol.export方法的参数

DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
Exporter<?> exporter = protocol.export(wrapperInvoker);

Protocol#getAdaptiveExtension

Protocol扩展也是类似，动态生成的扩展实现类源码如下

public class Protocol$Adaptive implements org.apache.dubbo.rpc.Protocol {
    private static final org.apache.dubbo.common.logger.Logger logger = org.apache.dubbo.common.logger.LoggerFactory.getLogger(ExtensionLoader.class);
    private java.util.concurrent.atomic.AtomicInteger count = new java.util.concurrent.atomic.AtomicInteger(0);

    public void destroy() {
        throw new UnsupportedOperationException("method public abstract void org.apache.dubbo.rpc.Protocol.destroy() of interface org.apache.dubbo.rpc.Protocol is not adaptive method!");
    }

    public int getDefaultPort() {
        throw new UnsupportedOperationException("method public abstract int org.apache.dubbo.rpc.Protocol.getDefaultPort() of interface org.apache.dubbo.rpc.Protocol is not adaptive method!");
    }

    public org.apache.dubbo.rpc.Invoker refer(java.lang.Class arg0, org.apache.dubbo.common.URL arg1) throws org.apache.dubbo.rpc.RpcException {
        if (arg1 == null) throw new IllegalArgumentException("url == null");
        org.apache.dubbo.common.URL url = arg1;
        String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(org.apache.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
        org.apache.dubbo.rpc.Protocol extension = null;
        try {
            extension = (org.apache.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.Protocol.class).getExtension(extName);
        } catch (Exception e) {
            if (count.incrementAndGet() == 1) {
                logger.warn("Failed to find extension named " + extName + " for type org.apache.dubbo.rpc.Protocol, will use default extension dubbo instead.", e);
            }
            extension = (org.apache.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.Protocol.class).getExtension("dubbo");
        }
        return extension.refer(arg0, arg1);
    }

    public org.apache.dubbo.rpc.Exporter export(org.apache.dubbo.rpc.Invoker arg0) throws org.apache.dubbo.rpc.RpcException {
        if (arg0 == null) throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null");
        if (arg0.getUrl() == null)
            throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument getUrl() == null");
        org.apache.dubbo.common.URL url = arg0.getUrl();
        String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(org.apache.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
        org.apache.dubbo.rpc.Protocol extension = null;
        try {
            extension = (org.apache.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.Protocol.class).getExtension(extName);
        } catch (Exception e) {
            if (count.incrementAndGet() == 1) {
                logger.warn("Failed to find extension named " + extName + " for type org.apache.dubbo.rpc.Protocol, will use default extension dubbo instead.", e);
            }
            extension = (org.apache.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.Protocol.class).getExtension("dubbo");
        }
        return extension.export(arg0);
    }
}

注意其中的export方法，即根据参数Invoker中的url信息来决定使用哪种Protocol。而这里的Invoker参数，就是前面的wrapperInvoker，wrapperInvoker中持有Invoker和ServiceConfig的信息，此时Invoker中的url信息如下

registry://localhost:2181/org.apache.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.2&export=dubbo://10.10.129.71:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.10.129.71&bind.port=20880&dubbo=2.0.2&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=11484&qos.port=22222&side=provider&timestamp=1543673781917&pid=11484&qos.port=22222&registry=zookeeper&timestamp=1543673781900

registry对应的Protocol是RegistryProtocol

registry=org.apache.dubbo.registry.integration.RegistryProtocol

所以，这里会调用RegistryProtocol的export方法，也就是说非injvm协议的服务都是通过RegistryProtocol进行发布。除此之外，Protocol也有对应的Wrapper扩展实现类：ProtocolFilterWrapper、ProtocolListenerWrapper。不过这两个Wrapper扩展实现类针对于Registry类型的Invoker并没有插入任何逻辑，如下：

// ProtocolFilterWrapper.java
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
    // Registry类型的Invoker不做处理
    if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
        return protocol.export(invoker);
    }
    return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
}

// ProtocolListenerWrapper.java
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
    // Registry类型的Invoker不做处理
    if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
        return protocol.export(invoker);
    }
    return new ListenerExporterWrapper<T>(protocol.export(invoker),
            Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class)
                    .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));
}

所以，此处会直接调用RegistryProtocol的export方法，RegistryProtocol负责注册服务到注册中心和向注册中心订阅服务，这一部分内容非常的多，需要重点介绍。

RegistryProtocol

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
    //export invoker，交给具体的协议去暴露服务，以dubbo协议为例子，
    final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

    URL registryUrl = getRegistryUrl(originInvoker);

    //registry provider
    final Registry registry = getRegistry(originInvoker);
    final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);

    //to judge to delay publish whether or not
    boolean register = registeredProviderUrl.getParameter("register", true);

    ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);

    if (register) {
        register(registryUrl, registeredProviderUrl);
        ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
    }

    // Subscribe the override data
    // FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call the same service. Because the subscribed is cached key with the name of the service, it causes the subscription information to cover.
    final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl);
    final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
    overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
    registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
    //Ensure that a new exporter instance is returned every time export
    return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl);
}

doLocalExport

非Registry类型的Invoker的导出过程，主要是将本地ip和20880端口打开，进行监听，最后包装成exporter返回。

// 缓存 key : exporter
private final Map<String, ExporterChangeableWrapper<?>> bounds = new ConcurrentHashMap<>();

private <T> ExporterChangeableWrapper<T>  doLocalExport(final Invoker<T> originInvoker) {
    String key = getCacheKey(originInvoker);
    ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
    if (exporter == null) {
        synchronized (bounds) {
            exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
            if (exporter == null) {
                // 得到一个Invoker代理，里面包含原来的Invoker
                final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
                // 调用 DubboProtocol的export方法，然后生成一个exporter放到缓存中
                exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
                bounds.put(key, exporter);
            }
        }
    }
    return exporter;
}

此处的export就是DubboProtocol，但是Protocol有两个Wrapper类：ProtocolListenerWrapper、ProtocolFilterWrapper。
ProtocolListenerWrapper#export

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
    // Registry类型的Invoker不做处理
    if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
        return protocol.export(invoker);
    }
    // 先导出为exporter，即 protocol.export(invoker)，如果有ExporterListener就用ExporterListener包装刚刚导出的expoter
    return new ListenerExporterWrapper<T>(protocol.export(invoker),
            Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class)
                    .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));
}

// ListenerExporterWrapper的构造方法如下
public ListenerExporterWrapper(Exporter<T> exporter, List<ExporterListener> listeners) {
    if (exporter == null) {
        throw new IllegalArgumentException("exporter == null");
    }
    this.exporter = exporter;
    this.listeners = listeners;
    if (listeners != null && !listeners.isEmpty()) {
        RuntimeException exception = null;
        for (ExporterListener listener : listeners) {
            if (listener != null) {
                try {
                    listener.exported(this);
                } catch (RuntimeException t) {
                    logger.error(t.getMessage(), t);
                    exception = t;
                }
            }
        }
        if (exception != null) {
            throw exception;
        }
    }
}

ProtocolFilterWrapper#export
ProtocolFilterWrapper的export方法和Filter有关，构建Filter链路，最后一个真正的Invoker

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));
}

DubboProtocol#export

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
    URL url = invoker.getUrl();

    // export service. key 由 port, serviceName, serviceVersion, serviceGroup 四部分组成
    String key = serviceKey(url);
    // 当nio客户端发起远程调用时，nio服务端通过此key来决定调用哪个Exporter，也就是执行的Invoker，因为exporter持有与其对应的invoker
    DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
    exporterMap.put(key, exporter);

    //export an stub service for dispatching event 下面这一段没看懂 todo
    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);
        }
    }

    // 建立NIO Server端
    openServer(url);
    // 序列化，对url序列化之后发布到注册中心
    optimizeSerialization(url);
    return exporter;
}

建立NIO Server端

private void openServer(URL url) {
    // find server. key是IP:PORT localhost:20880
    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);
        // 同一JVM中，同协议的服务，共享同一个Server，第一个暴露服务的时候创建Server，以后相同协议的服务都使用同一个Server
        if (server == null) {
            synchronized (this) {
                server = serverMap.get(key);
                if (server == null) {
                    serverMap.put(key, createServer(url));
                }
            }
        } else {
            // server supports reset, use together with override
            //accept、idleTimeout、threads、heartbeat参数的变化会引起Server的属性发生变化，这时需要重新设置Server
            server.reset(url);
        }
    }
}

紧接着调用createServer方法

private ExchangeServer createServer(URL url) {
    ...... 一些参数设置......
    ExchangeServer server;
    try {
        // Exchangers是门面类，里面封装的是Exchanger的逻辑，Exchanger默认只有一个实现HeaderExchanger。
        // Exchanger负责数据交换和网络通信，从Protocol进入Exchanger，标志着程序进入了remote层。
        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;
}

当执行Exchangers.bind方法的的时候，会创建一个实例，即会执行以下代码

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
    return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}

执行Transporters.bind方法的时候，会通过ExtensionLoader找到Transporter扩展的默认实现类，默认是NettyTransporter，这对应的是netty4，除此之外，还有一个neety3。NettyTransporter的bind方法如下：

public Server bind(URL url, ChannelHandler listener) throws RemotingException {
    return new NettyServer(url, listener);
}

// NettyServer,java
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
    super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}

// AbstractServer.java
public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
    super(url, handler);
    localAddress = getUrl().toInetSocketAddress();

    String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
    int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
    if (url.getParameter(Constants.ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
        bindIp = NetUtils.ANYHOST;
    }
    bindAddress = new InetSocketAddress(bindIp, bindPort);
    this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
    this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
    try {
        // 调用子类中的doOpen方法，这里是NettyServer
        doOpen();
    } catch (Throwable t) {
        throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()+ " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
    }
    //fixme replace this with better method 不懂
    DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
    executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort()));
}

在AbstractServer的构造函数中，又调用了子类的doOpen方法，这里是NettyServer。

protected void doOpen() throws Throwable {
    bootstrap = new ServerBootstrap();

    bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
    workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
            new DefaultThreadFactory("NettyServerWorker", true));

    final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
    channels = nettyServerHandler.getChannels();

    bootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
            .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .childHandler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
                    ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                            .addLast("decoder", adapter.getDecoder())
                            .addLast("encoder", adapter.getEncoder())
                            .addLast("handler", nettyServerHandler);
                }
            });
    // bind
    ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
    channelFuture.syncUninterruptibly();
    channel = channelFuture.channel();
}

终于看到和netty相关的API了，当服务端收到请求之后，先由NettyCodecAdapter解码，再由NettyHandler处理具体的业务逻辑，再由NettyCodecAdapter编码后发送。

getRegistryUrl

执行getRegistryUrl方法之前，即originInvoker中的url信息如下：

registry://localhost:2181/org.apache.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.2&export=dubbo://10.10.129.71:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.10.129.71&bind.port=20880&dubbo=2.0.2&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=14608&qos.port=22222&side=provider&timestamp=1543738199517&pid=14608&qos.port=22222&registry=zookeeper&timestamp=1543738199504

执行方法之后返回的URL信息如下：

zookeeper://localhost:2181/org.apache.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.2&export=dubbo://10.10.129.71:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.10.129.71&bind.port=20880&dubbo=2.0.2&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=14608&qos.port=22222&side=provider&timestamp=1543738199517&pid=14608&qos.port=22222&timestamp=1543738199504

getRegistry

private RegistryFactory registryFactory;

private Registry getRegistry(final Invoker<?> originInvoker) {
    // getRegistryUrl方法之前已经介绍过了
    URL registryUrl = getRegistryUrl(originInvoker);
    return registryFactory.getRegistry(registryUrl);
}

对于registryFactory这属性，我一开始非常不理解这是在什么时候被初始化的。后面想到RegistryProtocol和RegistryFactory都是SPI，而dubbo中的SPI是有IOC功能的，所以这个属性是在创建RegistryProtocol的时候自动注入进来的，即： RegistryFactory$Adaptive，动态生成的代码如下

package org.apache.dubbo.registry;

import org.apache.dubbo.common.extension.ExtensionLoader;

/**
 * @author spilledyear
 * @date 2018/12/2 16:30
 */
public class RegistryFactory$Adaptive implements org.apache.dubbo.registry.RegistryFactory {
    private static final org.apache.dubbo.common.logger.Logger logger = org.apache.dubbo.common.logger.LoggerFactory.getLogger(ExtensionLoader.class);
    private java.util.concurrent.atomic.AtomicInteger count = new java.util.concurrent.atomic.AtomicInteger(0);

    public org.apache.dubbo.registry.Registry getRegistry(org.apache.dubbo.common.URL arg0) {
        if (arg0 == null) throw new IllegalArgumentException("url == null");
        org.apache.dubbo.common.URL url = arg0;
        String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(org.apache.dubbo.registry.RegistryFactory) name from url(" + url.toString() + ") use keys([protocol])");
        org.apache.dubbo.registry.RegistryFactory extension = null;
        try {
            extension = (org.apache.dubbo.registry.RegistryFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.registry.RegistryFactory.class).getExtension(extName);
        } catch (Exception e) {
            if (count.incrementAndGet() == 1) {
                logger.warn("Failed to find extension named " + extName + " for type org.apache.dubbo.registry.RegistryFactory, will use default extension dubbo instead.", e);
            }
            extension = (org.apache.dubbo.registry.RegistryFactory) ExtensionLoader.getExtensionLoader(org.apache.dubbo.registry.RegistryFactory.class).getExtension("dubbo");
        }
        return extension.getRegistry(arg0);
    }
}

由上可知，最终获取到的是ZookeeperRegistryFactory，而ZookeeperRegistryFactory也是一AbstractRegistryFactory抽象类，getRegistry方法就是在AbstractRegistryFactory抽象类中实现的

public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
    private ZookeeperTransporter zookeeperTransporter;

    public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
        this.zookeeperTransporter = zookeeperTransporter;
    }

    @Override
    public Registry createRegistry(URL url) {
        return new ZookeeperRegistry(url, zookeeperTransporter);
    }

}

AbstractRegistryFactory的getRegistry方法实现如下：

public Registry getRegistry(URL url) {
    url = url.setPath(RegistryService.class.getName()).addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName()).removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
    String key = url.toServiceString();
    // Lock the registry access process to ensure a single instance of the registry
    LOCK.lock();
    try {
        Registry registry = REGISTRIES.get(key);
        if (registry != null) {
            return registry;
        }
        registry = createRegistry(url);
        if (registry == null) {
            throw new IllegalStateException("Can not create registry " + url);
        }
        REGISTRIES.put(key, registry);
        return registry;
    } finally {
        // Release the lock
        LOCK.unlock();
    }
}

在上述代码中，url.setPath方法执行之后的URL信息如下，主要就是把export相关的信息删掉

zookeeper://localhost:2181/org.apache.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=1336&qos.port=22222&timestamp=1543739882657

createRegistry方法就是创建一个ZookeeperRegistry对象，ZookeeperRegistry继承自FailbackRegistry类，FailbackRegistry又继承自AbstractRegistry抽象类，所以构造ZookeeperRegistry对象时代码执行如下：AbstractRegistry构造函数 =》 FailbackRegistry构造函数 =》 ZookeeperRegistry构造函数

AbstractRegistry

public AbstractRegistry(URL url) {
    setUrl(url);
    // Start file save timer
    // 配置中心的URL中是否配置了同步保存文件属性，否则默认为false
    syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
    // 配置信息本地缓存的文件名
    String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getParameter(Constants.APPLICATION_KEY) + "-" + url.getAddress() + ".cache");
    File file = null;
    if (ConfigUtils.isNotEmpty(filename)) {
        file = new File(filename);
        if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) {
            if (!file.getParentFile().mkdirs()) {
                throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
            }
        }
    }
    this.file = file;
    // 如果现有配置缓存，则从缓存文件中加载属性
    loadProperties();
    // 通知订阅，有关于这一步，不是很懂，之后再补充
    notify(url.getBackupUrls());
}

FailbackRegistry
这个主要和失败重试有关

public FailbackRegistry(URL url) {
    super(url);
    this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
    // 负责注册中心失效重试逻辑的
    this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
        @Override
        public void run() {
            // Check and connect to the registry
            try {
                retry();
            } catch (Throwable t) { // Defensive fault tolerance
                logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
            }
        }
    }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
}

ZookeeperRegistry

public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
    super(url);
    if (url.isAnyHost()) {
        throw new IllegalStateException("registry address == null");
    }
    // 获得到注册中心中的分组，默认dubbo
    String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
    if (!group.startsWith(Constants.PATH_SEPARATOR)) {
        group = Constants.PATH_SEPARATOR + group;
    }
    this.root = group;
    // ZookeeperTransport这里是生成的自适应实现，默认使用ZkClientZookeeperTransporter
    zkClient = zookeeperTransporter.connect(url);
    // ZkClientZookeeperClient添加状态改变监听器
    zkClient.addStateListener(new StateListener() {
        @Override
        public void stateChanged(int state) {
            if (state == RECONNECTED) {
                try {
                    recover();
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    });
}

getRegisteredProviderUrl

在RegistryProtocol中获取到Registry之后，Registry实例中保存着连接到了zookeeper的zkClient实例之后，下一步获取要注册到注册中心的url，即 getRegisteredProviderUrl 方法，执行该方法之后，得到的URL信息如下：

dubbo://10.10.129.71:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.2&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=400&side=provider&timestamp=1543743322671

ProviderConsumerRegTable.registerProvider

ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);

这是啥意思？不是很懂？

register

public void register(URL registryUrl, URL registedProviderUrl) {
    // 额，为什么这里又获取一次？什么操作？
    Registry registry = registryFactory.getRegistry(registryUrl);
    // 调用FailbackRegistry中的register方法
    registry.register(registedProviderUrl);
}

// FailbackRegistry.java
public void register(URL url) {
    // AbstractRegistry中缓存起来已经注册的url信息
    super.register(url);
    failedRegistered.remove(url);
    failedUnregistered.remove(url);
    try {
        // 调用子类的doRegister方法
        doRegister(url);
    } catch (Exception e) {
        Throwable t = e;

        // If the startup detection is opened, the Exception is thrown directly.
        boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                && url.getParameter(Constants.CHECK_KEY, true)
                && !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
        boolean skipFailback = t instanceof SkipFailbackWrapperException;
        if (check || skipFailback) {
            if (skipFailback) {
                t = t.getCause();
            }
            throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
        } else {
            logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
        }

        // Record a failed registration request to a failed list, retry regularly
        failedRegistered.add(url);
    }
}

// ZookeeperRegistry.java
protected void doRegister(URL url) {
    try {
        // 在zk中创建节点
        zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
    } catch (Throwable e) {
        throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

订阅注册中心的服务 todo

RegistryProtocol的export方法中剩下的几行代码

final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);

【DUBBO】服务暴露

ServiceBean实例化

实现DisposableBean

实现BeanNameAware

实现ApplicationContextAware

实现ApplicationListener

实现InitializingBean

服务暴露

本地暴露

远程暴露

ProxyFactory#getAdaptiveExtension

Protocol#getAdaptiveExtension

RegistryProtocol

doLocalExport

getRegistryUrl

getRegistry

getRegisteredProviderUrl

ProviderConsumerRegTable.registerProvider

register

订阅注册中心的服务 todo

猜你喜欢

热点阅读

【DUBBO】 服务暴露

ServiceBean实例化

实现DisposableBean

实现BeanNameAware

实现ApplicationContextAware

实现ApplicationListener

实现InitializingBean

服务暴露

本地暴露

远程暴露

ProxyFactory#getAdaptiveExtension

Protocol#getAdaptiveExtension

RegistryProtocol

doLocalExport

getRegistryUrl

getRegistry

getRegisteredProviderUrl

ProviderConsumerRegTable.registerProvider

register

订阅注册中心的服务 todo

猜你喜欢

热点阅读

【DUBBO】服务暴露