dubbo源码阅读之服务导出

dubbo服务导出

常见的使用dubbo的方式就是通过spring配置文件进行配置。例如下面这样

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"

       xsi:schemaLocation="http://www.springframework.org/schema/beans

       http://www.springframework.org/schema/beans/spring-beans.xsd

       http://dubbo.apache.org/schema/dubbo

       http://dubbo.apache.org/schema/dubbo/dubbo.xsd">

    <dubbo:application name="helloService-provider"/>

    <dubbo:registry address="zookeeper://localhost:2181"/>

    <dubbo:reference interface="com.zhuge.learn.dubbo.services.HelloService"

                     check="false" id="helloService">

        <dubbo:method name="sayHello" retries="2"/>

    </dubbo:reference>

</beans>

spring对于非默认命名空间的标签的解析是通过NamespaceHandlerResolver实现的，NamespaceHandlerResolver也算是一种SPI机制，通过解析jar包中的META-INF/spring.handlers文件，将所有的NamespaceHandler实现类以k-v的形式解析出来并放到内存中。所以要想扩展spring的命名空间，就要实现一个NamespaceHandler。

dubbo实现了自己的命名空间，对应的NamespaceHandler实现类是com.alibaba.dubbo.config.spring.schema.DubboNamespaceHandler。这个类也很简单，就是定义了用于解析不同标签的BeanDefinition解析类。但是dubbo的实现稍有不同，它将所有标签的解析都放到同一个类同一个方法中，个人认为这种设计欠妥，不利于扩展新的标签。

如果我们要创建一个服务提供者，我们需要在配置文件中配置service标签，所以dubbo的服务导出一定与这个标签相关。查看DubboNamespaceHandler代码会发现，服务导出的逻辑主要是由ServiceBean实现的，所以接下来我们就以ServiceBean为入口，一步步来分析dubbo的服务导出过程。

ServiceBean概览

ServiceBean继承了ServiceConfig类，同时实现了一大堆接口，这些接口基本上都与spring框架相关。其中ApplicationListener接口会监听ContextRefreshedEvent事件，这个事件是在spring容器完成刷新后发布的，导出逻辑的入口就在onApplicationEvent方法中。

onApplicationEvent

public void onApplicationEvent(ContextRefreshedEvent event) {

// 如果已经导出或者关闭服务，就忽略该事件

    if (!isExported() && !isUnexported()) {

        if (logger.isInfoEnabled()) {

            logger.info("The service ready on spring started. service: " + getInterface());

        }

        export();

    }

}

ServiceConfig.export

真正的导出服务的逻辑在父类方法中

// 这是一个同步方法，保证多线程情况下不会同时进行服务导出

public synchronized void export() {

    // 检查一些配置是否为空，对于空的配置创建默认的配置

    checkAndUpdateSubConfigs();

    if (!shouldExport()) {

        return;

    }

    if (shouldDelay()) {

        // 延迟导出服务

        delayExportExecutor.schedule(this::doExport, delay, TimeUnit.MILLISECONDS);

    } else {

        doExport();

    }

}

protected synchronized void doExport() {

        // 首先做一些状态检查

        // 如果已经反导出服务，说明服务已经被关闭

        if (unexported) {

            throw new IllegalStateException("The service " + interfaceClass.getName() + " has already unexported!");

        }

        // 如果已经导出过了，就不需要重复导出了

        if (exported) {

            return;

        }

        exported = true;

        // 如果服务名为空，以服务接口名作为服务名称

        if (StringUtils.isEmpty(path)) {

            path = interfaceName;

        }

        doExportUrls();

    }

doExportUrls

我们直接进入核心代码，

private void doExportUrls() {

// 加载所有的注册中心的URL

List registryURLs = loadRegistries(true);

// 如果配置了多个协议，那么每种协议都要导出，并且是对所有可用的注册url进行注册

for (ProtocolConfig protocolConfig : protocols) {

// 拼接服务名称，这里的path一般就是服务名

String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version);

// 服务提供者模型，用于全面描述服务提供者的信息

ProviderModel providerModel = new ProviderModel(pathKey, ref, interfaceClass);

ApplicationModel.initProviderModel(pathKey, providerModel);

// 导出这个服务提供者，

// 向所有的可用的注册中心进行注册

doExportUrlsFor1Protocol(protocolConfig, registryURLs);

}

}

从这段代码，我们可以看出来，dubbo会对配置的每个协议类型，每个注册中心全部进行服务导出和注册，服务导出和注册的次数=协议类型数*注册中心数

doExportUrlsFor1Protocol

这段代码主要封装了参数解析，和url拼装的逻辑。

创建代理类由ProxyFactory实现，

创建本地服务并注册到注册中心有RegistryProtocol实现

// 导出服务的核心代码

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {

    // 协议名称

    String name = protocolConfig.getName();

    // 如果没有配置协议名称，默认是dubbo

    if (StringUtils.isEmpty(name)) {

        name = Constants.DUBBO;

    }

    Map<String, String> map = new HashMap<String, String>();

    // 设置side属性为provider，side表示服务提供者还是消费者

    map.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE);

    // 添加运行时信息，包括

    // 1. dubbo协议的版本号，2.0.10 ~ 2.6.2

    // 2. dubbo版本号

    // 3. 时间戳 信息

    // 4. jvm进程号

    appendRuntimeParameters(map);

    // 添加ApplicationConfig, 配置属性

    appendParameters(map, application);

    // 添加ModuleConfig配置属性，模块配置，覆盖全局配置

    appendParameters(map, module);

    // 添加ProviderConfig配置属性

    appendParameters(map, provider, Constants.DEFAULT_KEY);

    // 添加协议配置，覆盖前面的配置

    appendParameters(map, protocolConfig);

    // 添加当前服务的配置，service标签的配置，覆盖前面的配置

    // 容易看出来，配置的优先级：service > protocol > provider > module > application

    appendParameters(map, this);

    if (CollectionUtils.isNotEmpty(methods)) {

        // 添加方法配置

        for (MethodConfig method : methods) {

            appendParameters(map, method, method.getName());

            // 替换retry配置

            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 (CollectionUtils.isNotEmpty(arguments)) {

                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)), ","));

        }

    }

    // 添加token信息

    if (!ConfigUtils.isEmpty(token)) {

        if (ConfigUtils.isDefault(token)) {

            map.put(Constants.TOKEN_KEY, UUID.randomUUID().toString());

        } else {

            map.put(Constants.TOKEN_KEY, token);

        }

    }

    // export service

    // 导出服务

    // 添加bind.ip属性，并返回用于注册的ip

    String host = this.findConfigedHosts(protocolConfig, registryURLs, map);

    // 添加bind.port属性，并返回用于注册的port

    Integer port = this.findConfigedPorts(protocolConfig, name, map);

    // 根据前面获取的参数信息创建一个URL

    URL url = new URL(name, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), map);

    // 对URL进行额外的配置

    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

    // scope属性值是none的不进行导出，直接忽略

    if (!Constants.SCOPE_NONE.equalsIgnoreCase(scope)) {

        // export to local if the config is not remote (export to remote only when config is remote)

        // 只要scope属性不等于remote就会进行本地导出

        if (!Constants.SCOPE_REMOTE.equalsIgnoreCase(scope)) {

            exportLocal(url);

        }

        // export to remote if the config is not local (export to local only when config is local)

        // 只要scope属性不等于local就会进行远程导出

        if (!Constants.SCOPE_LOCAL.equalsIgnoreCase(scope)) {

            if (logger.isInfoEnabled()) {

                logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);

            }

            if (CollectionUtils.isNotEmpty(registryURLs)) {

                // 对每一个注册中心都进行导出

                for (URL registryURL : registryURLs) {

                    // 添加dynamic属性的参数

                    url = url.addParameterIfAbsent(Constants.DYNAMIC_KEY, registryURL.getParameter(Constants.DYNAMIC_KEY));

                    // 加载监控中心的url，监控中心也是一个服务提供者

                    URL monitorUrl = loadMonitor(registryURL);

                    if (monitorUrl != null) {

                        // 添加参数到url中

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

                    }

                    // ref属性是通过spring容器的IOC特性自动注入的，

                    // 在DubboBeanDefinitionParser中对该属性进行了解析

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

                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

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

            }

            /**

             * @since 2.7.0

             * ServiceData Store

             */

            MetadataReportService metadataReportService = null;

            if ((metadataReportService = getMetadataReportService()) != null) {

                metadataReportService.publishProvider(url);

            }

        }

    }

    // 记录已经导出的

    this.urls.add(url);

}

ProxyFactory

@SPI("javassist")

public interface ProxyFactory {

    /**

     * create proxy.

     *

     * @param invoker

     * @return proxy

     */

    @Adaptive({Constants.PROXY_KEY})

    <T> T getProxy(Invoker<T> invoker) throws RpcException;

    /**

     * create proxy.

     *

     * @param invoker

     * @return proxy

     */

    @Adaptive({Constants.PROXY_KEY})

    <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException;

    /**

     * create invoker.

     *

     * @param <T>

     * @param proxy

     * @param type

     * @param url

     * @return invoker

     */

    // 这里规定了以proxy为key去url中查找扩展名，如果没有设置就用默认扩展名，

    // 默认扩展名是由SPI注解确定的，ProxyFactory的默认扩展名就是javassist

    // 查看META-INF/dubbo/internal/org.apache.dubbo.rpc.ProxyFactory文件，我们知道

    // javassist对应的扩展类就是org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory

    @Adaptive({Constants.PROXY_KEY})

    <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException;

}

所以默认实现类是JavassistProxyFactory

JavassistProxyFactory.getInvoker

@Override

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {

    // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'

    // Wrapper不能正确处理类名中带有$的情况

    // 获取一个包装类，用来根据传入的参数调用原始对象的不同方法

    // 起到的作用就是方法路由。

    // jdk动态代理使用反射调用不同的方法，效率较低。

    // 而javaassist通过方法名以及参数个数和参数类型进行判断具体调用哪个方法，效率更高

    final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);

    // 生成一个Invoker，内部仅仅是调用wrapper.invokeMethod方法

    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.makeWrapper

真正负责代码生成的是Wrapper.makeWrapper方法。这段代码比较长，逻辑比较复杂，而且代码生成的逻辑又很繁琐，其实也没有什么高深的技术，所以我决定直接用单元测试来生成一段代码，这样就能直观第理解生成的代码长什么样

首先给出一个被代理的接口

public interface I2 {

    void setName(String name);

    void hello(String name);

    int showInt(int v);

    float getFloat();

    void setFloat(float f);

}

下面就是Wrapper.makeWrapper方法最后生成的代码的样子，

public class Wrapper0 extends Wrapper {

    public static String[] pns;

    public static java.util.Map pts;

    public static String[] mns;

    public static String[] dmns;

    public static Class[] mts0;

    public static Class[] mts1;

    public static Class[] mts2;

    public static Class[] mts3;

    public static Class[] mts4;

    public static Class[] mts5;

    public String[] getPropertyNames() {

        return pns;

    }

    public boolean hasProperty(String n) {

        return pts.containsKey($1);

    }

    public Class getPropertyType(String n) {

        return (Class) pts.get($1);

    }

    public String[] getMethodNames() {

        return mns;

    }

    public String[] getDeclaredMethodNames() {

        return dmns;

    }

    public void setPropertyValue(Object o, String n, Object v) {

        org.apache.dubbo.common.bytecode.I2 w;

        try {

            w = ((org.apache.dubbo.common.bytecode.I2) $1);

        } catch (Throwable e) {

            throw new IllegalArgumentException(e);

        }

        if ($2.equals("name")) {

            w.setName((java.lang.String) $3);

            return;

        }

        if ($2.equals("float")) {

            w.setFloat(((Number) $3).floatValue());

            return;

        }

        throw new org.apache.dubbo.common.bytecode.NoSuchPropertyException("Not found property \"" + $2 + "\" field or setter method in class org.apache.dubbo.common.bytecode.I2.");

    }

    public Object getPropertyValue(Object o, String n) {

        org.apache.dubbo.common.bytecode.I2 w;

        try {

            w = ((org.apache.dubbo.common.bytecode.I2) $1);

        } catch (Throwable e) {

            throw new IllegalArgumentException(e);

        }

        if ($2.equals("float")) {

            return ($w) w.getFloat();

        }

        if ($2.equals("name")) {

            return ($w) w.getName();

        }

        throw new org.apache.dubbo.common.bytecode.NoSuchPropertyException("Not found property \"" + $2 + "\" field or setter method in class org.apache.dubbo.common.bytecode.I2.");

    }

    public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws java.lang.reflect.InvocationTargetException {

        org.apache.dubbo.common.bytecode.I2 w;

        try {

            w = ((org.apache.dubbo.common.bytecode.I2) $1);

        } catch (Throwable e) {

            throw new IllegalArgumentException(e);

        }

        try {

            if ("getFloat".equals($2) && $3.length == 0) {

                return ($w) w.getFloat();

            }

            if ("setName".equals($2) && $3.length == 1) {

                w.setName((java.lang.String) $4[0]);

                return null;

            }

            if ("setFloat".equals($2) && $3.length == 1) {

                w.setFloat(((Number) $4[0]).floatValue());

                return null;

            }

            if ("hello".equals($2) && $3.length == 1) {

                w.hello((java.lang.String) $4[0]);

                return null;

            }

            if ("showInt".equals($2) && $3.length == 1) {

                return ($w) w.showInt(((Number) $4[0]).intValue());

            }

            if ("getName".equals($2) && $3.length == 0) {

                return ($w) w.getName();

            }

        } catch (Throwable e) {

            throw new java.lang.reflect.InvocationTargetException(e);

        }

        throw new org.apache.dubbo.common.bytecode.NoSuchMethodException("Not found method \"" + $2 + "\" in class org.apache.dubbo.common.bytecode.I2.");

    }

}

其中方法中的参数用$1,$2这种形式表示，猜测在javassist中会进行处理。

我们主要看invokeMethod，逻辑相对还是很明了的，通过方法名和参数个数判断应该调用哪个方法。

到这里，Invoker对象就创建完成了，接下来就进入到服务导出的部分。

Protocol.export

@Adaptive

<T> Exporter<T> export(Invoker<T> invoker) throws RpcException;

这个方法带有Adaptive注解，是一个自适应方法，自适应扩展类，我们之前分析过，通过入参获取URL，通过URL获取指定key的值，用这个获取到的值作为扩展名加载扩展类，然后调用这个扩展类的方法。

但是export方法上的注解并没有给出key，回想一下生成自适应扩展类代码的细节，当Adaptive注解未指定key时，将接口名转换为key，Protocol会被转换为protocol，而对于key为protocol的情况会直接调用URL.getProtocol方法获取协议类型作为扩展名。

在loadRegistries方法中加载注册url时，已经将url的protocol属性设为registry，也就是说会使用org.apache.dubbo.registry.integration.RegistryProtocol来进行服务导出，接下来我们就来分析这个类。

RegistryProtocol.export

所以接下来我们就分析一下RegistryProtocol.export的导出过程

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {

    // 获取注册的url，并将protocol替换为相应的协议类型

    // 前面讲本来的协议类型设置到registry参数中，而将protocol参数设置为registry，

    // 这样做是为了在自适应扩展机制在查找扩展名时能够根据扩展名是registry找到RegistryProtocol

    // 找到之后并且进入这个类的方法之后，自然需要再把协议类型设置回来

    URL registryUrl = getRegistryUrl(originInvoker);

    // url to export locally

    // 获取服务提供者url，用于导出到本地

    // 注册的url中的一个参数，即export参数的值

    URL providerUrl = getProviderUrl(originInvoker);

    // 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.

    // 获取订阅URL，用于

    final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);

    // 创建监听器

    final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);

    overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);

    providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);

    //export invoker

    // 导出服务到本地

    final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);

    // url to registry

    final Registry registry = getRegistry(originInvoker);

    // 获取用于发送到注册中心的提供者url

    final URL registeredProviderUrl = getRegisteredProviderUrl(providerUrl, registryUrl);

    // 想服务提供者与消费者注册表中注册服务

    ProviderInvokerWrapper<T> providerInvokerWrapper = ProviderConsumerRegTable.registerProvider(originInvoker,

            registryUrl, registeredProviderUrl);

    //to judge if we need to delay publish

    boolean register = registeredProviderUrl.getParameter("register", true);

    if (register) {

        // 向注册中心注册

        register(registryUrl, registeredProviderUrl);

        providerInvokerWrapper.setReg(true);

    }

    // Deprecated! Subscribe to override rules in 2.6.x or before.

    registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);

    exporter.setRegisterUrl(registeredProviderUrl);

    exporter.setSubscribeUrl(overrideSubscribeUrl);

    //Ensure that a new exporter instance is returned every time export

    return new DestroyableExporter<>(exporter);

}

我们略过一些不太重要的，这个方法主要就做了两件事：

对提供者url和注册url进行处理
将服务导出到本地
向注册中心发送服务提供者信息

doLocalExport

private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {

    // 用于缓存的key,即提供者url

    String key = getCacheKey(originInvoker);

    // 如果服务在缓存中不存在，则需要进行导出

    return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(key, s -> {

        Invoker<?> invokerDelegete = new InvokerDelegate<>(originInvoker, providerUrl);

        // protocol成员变量在加载扩展类的时候会进行注入，通过SPI或spring容器查找到对应的

        // 通过SPI注入时会注入自适应扩展类，通过传入的url动态决定使用哪个Protocol

        return new ExporterChangeableWrapper<>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);

    });

}

就像注释中说的，通过自适应机制，根据运行时传入的Invoker中的url动态决定使用哪个Protocol，以常用的dubbo协议为例，对应的实现类是org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol

DubboProtocol.export

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {

    URL url = invoker.getUrl();

    // export service.

    // 服务的key组成：serviceGroup/serviceName:serviceVersion:port

    String key = serviceKey(url);

    // 创建一个DubboExporter。用于封装一些引用

    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;

}

主要的逻辑在openServer中

openServer

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) {

            synchronized (this) {

                server = serverMap.get(key);

                if (server == null) {

                    serverMap.put(key, createServer(url));

                }

            }

        } else {

            // server supports reset, use together with override

            server.reset(url);

        }

    }

}

这个方法的主要作用就是缓存的双重检查锁，创建的服务的代码在createServer中

createServer

终于到正题了

private ExchangeServer createServer(URL url) {

    url = URLBuilder.from(url)

            // send readonly event when server closes, it's enabled by default

            // 服务端关闭时发送只读事件

            .addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())

            // enable heartbeat by default

            // 设置心跳间隔

            .addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT))

            // 设置编码器

            .addParameter(Constants.CODEC_KEY, DubboCodec.NAME)

            .build();

    // 传输协议，默认是netty

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

    }

    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;

}

设置一些参数，如服务端关闭时发送只读事件，心跳间隔，编解码器等
绑定端口，启动服务

Exchangers.bind

public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {

    if (url == null) {

        throw new IllegalArgumentException("url == null");

    }

    if (handler == null) {

        throw new IllegalArgumentException("handler == null");

    }

    // 编解码器设为exchange

    url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");

    return getExchanger(url).bind(url, handler);

}

默认的Exchanger是HeaderExchanger，

HeaderExchanger.bind

@Override

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {

    return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));

}

服务启动逻辑在Transporters.bind中

Transporters.bind

public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {

    if (url == null) {

        throw new IllegalArgumentException("url == null");

    }

    if (handlers == null || handlers.length == 0) {

        throw new IllegalArgumentException("handlers == null");

    }

    ChannelHandler handler;

    if (handlers.length == 1) {

        handler = handlers[0];

    } else {

        handler = new ChannelHandlerDispatcher(handlers);

    }

    return getTransporter().bind(url, handler);

}

getTransporter方法返回的是自适应扩展类，会根据url决定使用哪个扩展类。

Transporter

@SPI("netty")

public interface Transporter {

    /**

     * Bind a server.

     *

     * @param url     server url

     * @param handler

     * @return server

     * @throws RemotingException

     * @see org.apache.dubbo.remoting.Transporters#bind(URL, ChannelHandler...)

     */

    // 依次根据server和transporter参数值决定扩展名

    @Adaptive({Constants.SERVER_KEY, Constants.TRANSPORTER_KEY})

    Server bind(URL url, ChannelHandler handler) throws RemotingException;

    /**

     * Connect to a server.

     *

     * @param url     server url

     * @param handler

     * @return client

     * @throws RemotingException

     * @see org.apache.dubbo.remoting.Transporters#connect(URL, ChannelHandler...)

     */

    @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})

    Client connect(URL url, ChannelHandler handler) throws RemotingException;

}

server参数的默认值是netty，所以我们分析一下NettyTransporter

NettyTransporter.bind

@Override

public Server bind(URL url, ChannelHandler listener) throws RemotingException {

    return new NettyServer(url, listener);

}

分析NettyServer构造器

NettyServer

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 = Constants.ANYHOST_VALUE;

    }

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

        if (logger.isInfoEnabled()) {

            logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());

        }

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

}

主要逻辑：设置参数，然后打开服务。doOpen方法由子类实现。

NettyServer.doOpen

protected void doOpen() throws Throwable {

    NettyHelper.setNettyLoggerFactory();

    // boss线程池，这里使用了newCachedThreadPool，如果需要就会创建新的线程，

    ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));

    // worker线程池

    ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));

    // 最大32核

    ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));

    // netty启动类

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

}

这里主要涉及到netty api的使用。设置boss线程池和worker线程池，然后通过启动类ServerBootstrap绑定指定的host和port上，开始监听端口。

服务启动到这里就告一段落，netty的部分就不再展开，不属于dubbo框架的内容。

我只想说，这代码也太特么的深了。最初看到dubbo的架构图，就是那张十层的图时，我不太理解，看完这些代码我才明白，dubbo为什么能把层分得那么细，那么清晰。

代码中基本上能抽象成接口的都抽象出来，扩展性是大大增强了，但是要想弄明白框架的整体架构就得多花点时间消化消化了。

另外，差点忘了一个重要内容，那就是netty的事件处理器，其实通过前面的层层调用我们可以发现，处理器类最开始在DubboProtocol被创建，沿着调用链一直传递到netty api。

DubboProtocol.requestHandler

这里面一个重要的方法就是reply方法，这个方法的主要内容就是检查参数类型，检查方法存不存在，然后调用原始的Invoker。

实际上从socket接收到字节数组怎么被解析为Invocation，这中间还有很长的调用链，通过代理模式进行层层封装，这块逻辑还不太懂，留着以后慢慢研究。

接下来，我们回过头再来分析注册的逻辑，也就是向注册中心发送服务提供者信息。这部分的入口在

服务注册

服务注册部分的逻辑不是很复杂，主要还是通过url中的protocol参数值通过自适应机制找到对应的RegistryFactory类，然后获取对应的Registry类。

以zookeeper为例，其实就是在zookeeper上创建对应的路径。当然不仅仅是这么简单，其中还有重试，失败回退等逻辑，这里不再细说，目的就是知道大概的原理。

巴特西