7.4 服務遠程暴露 - 創建Exporter與啟動netty服務端

為了安全：服務啟動的ip全部使用10.10.10.10

遠程服務的暴露總體步驟：

• 將ref封裝為invoker
• 將invoker轉換為exporter
• 啟動netty
• 注冊服務到zookeeper
• 訂閱
• 返回新的exporter實例

服務遠程暴露的代碼：

            //如果配置不是local則暴露為遠程服務.(配置為local，則表示只暴露本地服務)
            if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {
                if (logger.isInfoEnabled()) {
                    logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                }
                if (registryURLs != null && registryURLs.size() > 0
                        && url.getParameter("register", true)) {
                    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());
                        }
                        if (logger.isInfoEnabled()) {
                            logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                        }
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                } else {
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
                    Exporter<?> exporter = protocol.export(invoker);
                    exporters.add(exporter);
                }
            }

首先將實現類ref封裝為Invoker，之后將invoker轉換為exporter，最后將exporter放入緩存List<Exporter> exporters中。

 

一 將實現類ref封裝為Invoker

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

1  為registryURL拼接export=providerUrl參數

一開始的registryURL：

registry://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&pid=887&registry=zookeeper&timestamp=1507096022072

registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())這句代碼為registryURL添加了參數並編碼：（這里給出沒有編碼的樣子）

export=dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=887&side=provider&timestamp=1507096024334

2  ProxyFactory$Adaptive.getInvoker(DemoServiceImpl實例, Class<DemoService>, registryURL)

    public com.alibaba.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, java.lang.Class arg1, com.alibaba.dubbo.common.URL arg2) throws com.alibaba.dubbo.rpc.RpcException {
        if (arg2 == null)
            throw new IllegalArgumentException("url == null");
        com.alibaba.dubbo.common.URL url = arg2;
        String extName = url.getParameter("proxy", "javassist");//結果是javassist
        if(extName == null)
            throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
        com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        return extension.getInvoker(arg0, arg1, arg2);
    }

這里，本來是調用JavassistProxyFactory的getInvoker方法，但是JavassistProxyFactory被StubProxyFactoryWrapper給aop了。

3  StubProxyFactoryWrapper.getInvoker(DemoServiceImpl實例, Class<DemoService>, registryURL)

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

4  JavassistProxyFactory.getInvoker(DemoServiceImpl實例, Class<DemoService>, registryURL)

    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類：Wrapper.getWrapper(Class<DemoServiceImpl>)。該類記錄了DemoServiceImpl的屬性名稱，方法名稱等信息。關鍵代碼如下：（完整代碼見：7.2 服務本地暴露）

import com.alibaba.dubbo.common.bytecode.Wrapper;
import java.util.HashMap;

public class Wrapper1 extends Wrapper {

    public static String[] pns;//property name array
    public static java.util.Map pts = new HashMap();//<property key, property value>
    public static String[] mns;//method names
    public static String[] dmns;//
    public static Class[] mts0;
    /**
     * @param o  實現類
     * @param n  方法名稱
     * @param p  參數類型
     * @param v  參數名稱
     * @return
     * @throws java.lang.reflect.InvocationTargetException
     */
    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) o);
        } catch (Throwable e) {
            throw new IllegalArgumentException(e);
        }
        try {
            if ("sayHello".equals(n) && p.length == 1) {
                return ($w) w.sayHello((java.lang.String) v[0]);
            }
        } catch (Throwable e) {
            throw new java.lang.reflect.InvocationTargetException(e);
        }
        throw new com.alibaba.dubbo.common.bytecode.NoSuchMethodException("Not found method \"" + n + "\" in class com.alibaba.dubbo.demo.provider.DemoServiceImpl.");
    }
}

創建完DemoServiceImpl的Wrapper類之后（實際上該實例在本地暴露的時候已經存入緩存了，這里只是從緩存中拿出來而已），創建一個AbstractProxyInvoker實例。

    private final T proxy;
    private final Class<T> type;
    private final URL url;

    public AbstractProxyInvoker(T proxy, Class<T> type, URL url) {
        if (proxy == null) {
            throw new IllegalArgumentException("proxy == null");
        }
        if (type == null) {
            throw new IllegalArgumentException("interface == null");
        }
        if (!type.isInstance(proxy)) {
            throw new IllegalArgumentException(proxy.getClass().getName() + " not implement interface " + type);
        }
        this.proxy = proxy;
        this.type = type;
        this.url = url;
    }

最后創建完成的AbstractProxyInvoker實例屬性如下：

• proxy：DemoServiceImpl實例
• type：Class<com.alibaba.dubbo.demo.DemoService>
• url：registry://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&export=dubbo%3A%2F%2F10.10.10.10%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D993%26side%3Dprovider%26timestamp%3D1507100322516&pid=993&registry=zookeeper&timestamp=1507100319830

這樣我們就將ref實現類轉換成了Invoker，之后在調用該invoker.invoke(Invocation invocation)的時候，會調用invoker.doInvoke(T proxy, String methodName,Class<?>[] parameterTypes, Object[] arguments)的時候，就會調用相應的實現類proxy的wrapper類的invokeMethod(proxy, methodName, parameterTypes, arguments)，該方法又會調用真實的實現類methodName方法。這里可以先給出AbstractProxyInvoker.invoke(Invocation invocation)源碼：

    public Result invoke(Invocation invocation) throws RpcException {
        try {
            return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
        } catch (InvocationTargetException e) {
            return new RpcResult(e.getTargetException());
        } catch (Throwable e) {
            throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

這里的proxy就是上邊賦好值的proxy：DemoServiceImpl實例。而方法信息會封裝在Invocation對象中，該對象在服務引用時介紹。

 

二  將Invoker轉換為Exporter

Exporter<?> exporter = protocol.export(invoker)

1  Protocol$Adaptive.export(com.alibaba.dubbo.rpc.Invoker AbstractProxyInvoker實例)

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

這里，由於aop的原因，首先調用了ProtocolListenerWrapper的export(Invoker<T> invoker)，如下：

    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        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)));
    }

由於協議是“registry”，所以不做任何處理，繼續調用ProtocolFilterWrapper的export(Invoker<T> 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));
    }

同理，由於協議是“registry”，所以不做任何處理，繼續調用RegistryProtocol.export(final Invoker<T> originInvoker)，如下：

    public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        //export invoker
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        //registry provider
        final Registry registry = getRegistry(originInvoker);
        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
        registry.register(registedProviderUrl);
        // 訂閱override數據
        // FIXME 提供者訂閱時，會影響同一JVM即暴露服務，又引用同一服務的的場景，因為subscribed以服務名為緩存的key，導致訂閱信息覆蓋。
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        //保證每次export都返回一個新的exporter實例
        return new Exporter<T>() {
            public Invoker<T> getInvoker() {
                return exporter.getInvoker();
            }

            public void unexport() {
                try {
                    exporter.unexport();
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    registry.unregister(registedProviderUrl);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    overrideListeners.remove(overrideSubscribeUrl);
                    registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
            }
        };
    }

該方法完成了遠程暴露的全部流程。

• 將invoker轉換為exporter
• 啟動netty
• 注冊服務到zookeeper
• 訂閱
• 返回新的exporter實例

2  將invoker轉換為exporter並啟動netty服務

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

doLocalExport(final Invoker<T> originInvoker)

    /**
     * 1 從invoker的URL中的Map<String, String> parameters中獲取key為export的地址providerUrl，該地址將是服務注冊在zk上的節點
     * 2 從 Map<String, ExporterChangeableWrapper<?>> bounds 緩存中獲取key為上述providerUrl的exporter，如果有，直接返回，如果沒有，創建並返回
     * @return
     */
    @SuppressWarnings("unchecked")
    private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
        String key = getCacheKey(originInvoker);//根據originInvoker獲取providerUrl
        ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
        if (exporter == null) {
            synchronized (bounds) {
                exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
                if (exporter == null) {
                    final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));//存儲originInvoker和providerUrl
                    exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
                    bounds.put(key, exporter);
                }
            }
        }
        return exporter;
    }

2.1 從originInvoker中獲取providerUrl

該方法直接首先調用getCacheKey(final Invoker<?> originInvoker)中獲取providerUrl，這里的originInvoker就是上述創建出來的AbstractProxyInvoker實例，注意他的url是registry協議的，該url的export參數的value就是我們要獲取的providerUrl。獲取providerUrl的源碼如下：

    private String getCacheKey(final Invoker<?> originInvoker) {
        URL providerUrl = getProviderUrl(originInvoker);
        String key = providerUrl.removeParameters("dynamic", "enabled").toFullString();
        return key;
    }

    private URL getProviderUrl(final Invoker<?> origininvoker) {
        String export = origininvoker.getUrl().getParameterAndDecoded(Constants.EXPORT_KEY);
        if (export == null || export.length() == 0) {
            throw new IllegalArgumentException("The registry export url is null! registry: " + origininvoker.getUrl());
        }

        URL providerUrl = URL.valueOf(export);
        return providerUrl;
    }

之后一系列的操作，就是獲取該providerUrl對應的exporter，之后放入緩存Map<String, ExporterChangeableWrapper<?>> bounds中，所以一個providerUrl只會對應一個exporter。

2.2  創建InvokerDelegete

final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));

InvokerDelegete是RegistryProtocol的一個靜態內部類，該類是一個originInvoker的委托類，該類存儲了originInvoker，其父類InvokerWrapper還會存儲providerUrl，InvokerWrapper會調用originInvoker的invoke方法，也會銷毀invoker。可以管理invoker的生命周期。

    public static class InvokerDelegete<T> extends InvokerWrapper<T> {
        private final Invoker<T> invoker;

        /**
         * @param invoker
         * @param url     invoker.getUrl返回此值
         */
        public InvokerDelegete(Invoker<T> invoker, URL url) {
            super(invoker, url);
            this.invoker = invoker;
        }

        public Invoker<T> getInvoker() {
            if (invoker instanceof InvokerDelegete) {
                return ((InvokerDelegete<T>) invoker).getInvoker();
            } else {
                return invoker;
            }
        }
    }

InvokerWrapper的核心代碼：

public class InvokerWrapper<T> implements Invoker<T> {
    private final Invoker<T> invoker;//originInvoker
    private final URL url;//providerUrl

    public InvokerWrapper(Invoker<T> invoker, URL url) {
        this.invoker = invoker;
        this.url = url;
    }

    public boolean isAvailable() {
        return invoker.isAvailable();
    }

    public Result invoke(Invocation invocation) throws RpcException {
        return invoker.invoke(invocation);
    }

    public void destroy() {
        invoker.destroy();
    }
}

這樣一個InvokerDelegete對象就創建好了，屬性如下：

• invoker：originInvoker（AbstractProxyInvoker對象）
• InvokerWrapper.invoker：originInvoker（AbstractProxyInvoker對象）
• url：providerUrl（dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=1035&side=provider&timestamp=1507101286063）

2.3  使用DubboProtocol將InvokerDelegete轉換為Exporter

exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker)

2.3.1  Protocol$Adaptive.export(com.alibaba.dubbo.rpc.Invoker InvokerDelegete對象)

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

該代碼再貼最后一遍了。之后調用ProtocolListenerWrapper的ProtocolListenerWrapper.export(Invoker<T> InvokerDelegete)，之后調用ProtocolFilterWrapper.export(Invoker<T> InvokerDelegete)：首先對InvokerDelegete對象進行8個filter的遞歸包裝，之后使用DubboProtocol對包裝后的InvokerDelegete對象進行export。

層層包裝的源碼：

    /**
     * 1 根據key從url中獲取相應的filter的values，再根據這個values和group去獲取類上帶有@Active注解的filter集合
     * 2 之后將這些filter對傳入的invoker進行遞歸包裝層invoker（就是一個鏈表）
     */
    private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
        Invoker<T> last = invoker;
        List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
        if (filters.size() > 0) {
            for (int i = filters.size() - 1; i >= 0; i--) {
                final Filter filter = filters.get(i);
                final Invoker<T> next = last;
                last = new Invoker<T>() {

                    public Class<T> getInterface() {
                        return invoker.getInterface();
                    }

                    public URL getUrl() {
                        return invoker.getUrl();
                    }

                    public boolean isAvailable() {
                        return invoker.isAvailable();
                    }

                    public Result invoke(Invocation invocation) throws RpcException {
                        return filter.invoke(next, invocation);
                    }

                    public void destroy() {
                        invoker.destroy();
                    }

                    @Override
                    public String toString() {
                        return invoker.toString();
                    }
                };
            }
        }
        return last;
    }

上述方法中最重要的就是Invoker的Result invoke(Invocation invocation)，在該方法中，是使用了filter.invoke(next, invocation)，而這里的next又可能是另一個filter。這里我們打開一個filter來看一下源碼：

@Activate(group = Constants.PROVIDER, order = -110000)
public class EchoFilter implements Filter {
    public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
        if (inv.getMethodName().equals(Constants.$ECHO) && inv.getArguments() != null && inv.getArguments().length == 1)
            return new RpcResult(inv.getArguments()[0]);
        return invoker.invoke(inv);
    }
}

可以看到，該filter會調用傳入的next的invoke方法。

這里給出被遞歸包裝后的對象：（命名為InvokerDelegete的filter對象）

EchoFilter
-->ClassLoaderFilter
   -->GenericFilter
      -->ContextFilter
         -->TraceFilter
            -->TimeoutFilter
               -->MonitorFilter
                  -->ExceptionFilter
                     -->InvokerDelegete對象

2.3.2  DubboProtocol.export(Invoker<T> InvokerDelegete的filter對象)

    /**
     * 1 從invoker的url中獲取將要暴露的遠程服務的key：com.alibaba.dubbo.demo.DemoService:20880（實際上是：serviceGroup/serviceName:serviceVersion:port）
     * 注意：本地暴露的key就是：com.alibaba.dubbo.demo.DemoService
     * 2 打開ExchangeServer
     */
    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();
        // export service.
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        exporterMap.put(key, exporter);

        //export an stub service for dispaching 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);

        return exporter;
    }

首先從“InvokerDelegete的filter對象”中的url獲取key，這段代碼很簡單，就是獲取serviceGroup/serviceName:serviceVersion:port這樣形式的一個key，這里最后獲取到的是com.alibaba.dubbo.demo.DemoService:20880。

之后創建DubboExporter。

2.3.2.1 DubboExporter<T>(InvokerDelegete的filter對象, "com.alibaba.dubbo.demo.DemoService:20880", exporterMap)

public class DubboExporter<T> extends AbstractExporter<T> {
    //serviceGroup/serviceName:serviceVersion:port, 例如：com.alibaba.dubbo.demo.DemoService:20880
    private final String key;//
    //{ "com.alibaba.dubbo.demo.DemoService:20880" -> 當前的DubboExporter實例 }
    private final Map<String, Exporter<?>> exporterMap;

    public DubboExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap) {
        super(invoker);
        this.key = key;
        this.exporterMap = exporterMap;
    }

    @Override
    public void unexport() {
        super.unexport();
        exporterMap.remove(key);
    }
}

注意這里的exporterMap是引用傳遞。

父類：

public abstract class AbstractExporter<T> implements Exporter<T> {
    protected final Logger logger = LoggerFactory.getLogger(getClass());
    private final Invoker<T> invoker;
    private volatile boolean unexported = false;

    public AbstractExporter(Invoker<T> invoker) {
        if (invoker == null)
            throw new IllegalStateException("service invoker == null");
        if (invoker.getInterface() == null)
            throw new IllegalStateException("service type == null");
        if (invoker.getUrl() == null)
            throw new IllegalStateException("service url == null");
        this.invoker = invoker;
    }

    public Invoker<T> getInvoker() {
        return invoker;
    }

    public void unexport() {
        if (unexported) {
            return;
        }
        unexported = true;
        getInvoker().destroy();
    }

    public String toString() {
        return getInvoker().toString();
    }
}

這里，我們把一個“InvokerDelegete的filter對象”賦給了AbstractExporter的Invoker引用，也就是說從exporter中可以獲取到invoker。最后在DubboProtocol.export(Invoker<T> invoker)中執行：exporterMap.put(key, exporter); 這樣就將{ "com.alibaba.dubbo.demo.DemoService:20880" -> 當前的DubboExporter實例 }存儲起來了。

來看一下現在的DubboExporter實例：

• key：com.alibaba.dubbo.demo.DemoService:20880
• invoker：“InvokerDelegete的filter對象”
• exporterMap：{ "com.alibaba.dubbo.demo.DemoService:20880" -> 當前的DubboExporter實例 }

2.3.2.2 開啟ExchangeServer

    /**
     * 從緩存Map<String, ExchangeServer> serverMap中根據"host:port"獲取ExchangeServer，如果沒有，創建ExchangeServer，之后放入緩存。
     * @param url
     */
    private void openServer(URL url) {
        // find server.
        String key = url.getAddress();//10.10.10.10:20880
        //client 也可以暴露一個只有server可以調用的服務。
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                serverMap.put(key, createServer(url));
            } else {
                //server支持reset,配合override功能使用
                server.reset(url);
            }
        }
    }

首先從provderUrl中獲取host:port作為key，之后從緩存serverMap中獲取ExchangeServer，如果沒有，創建ExchangeServer，最后以如下方式放入緩存：

Map<String, ExchangeServer> serverMap：{ "10.10.10.10:20880"<->ExchangeServer實例 }。

創建ExchangeServer：createServer(URL providerUrl)

    private ExchangeServer createServer(URL url) {
        //默認開啟server關閉時發送readonly事件
        url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
        //默認開啟heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
        String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);

        if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);

        url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
        ExchangeServer server;
        try {
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        str = url.getParameter(Constants.CLIENT_KEY);
        if (str != null && str.length() > 0) {
            Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
            if (!supportedTypes.contains(str)) {
                throw new RpcException("Unsupported client type: " + str);
            }
        }
        return server;
    }

首先是在原本providerUrl上添加參數：channel.readonly.sent=true&heartbeat=60000&codec=dubbo（其中的heartbeat參數會在HeaderExchangeServer啟動心跳計時器時使用）

之后使用Exchangers.bind("添加參數后的providerUrl", requestHandler)創建ExchangeServer。首先來看一下DubboProtocol#requestHandler。這個類極其重要，后續經過層層包裝后，會成為最終netty的服務端邏輯處理器。

    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
        public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                Invoker<?> invoker = getInvoker(channel, inv);
                //如果是callback 需要處理高版本調用低版本的問題
                if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
                    String methodsStr = invoker.getUrl().getParameters().get("methods");
                    boolean hasMethod = false;
                    if (methodsStr == null || methodsStr.indexOf(",") == -1) {
                        hasMethod = inv.getMethodName().equals(methodsStr);
                    } else {
                        String[] methods = methodsStr.split(",");
                        for (String method : methods) {
                            if (inv.getMethodName().equals(method)) {
                                hasMethod = true;
                                break;
                            }
                        }
                    }
                    if (!hasMethod) {
                        logger.warn(new IllegalStateException("The methodName " + inv.getMethodName() + " not found in callback service interface ,invoke will be ignored. please update the api interface. url is:" + invoker.getUrl()) + " ,invocation is :" + inv);
                        return null;
                    }
                }
                RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                return invoker.invoke(inv);
            }
            throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
        }

        @Override
        public void received(Channel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                reply((ExchangeChannel) channel, message);
            } else {
                super.received(channel, message);
            }
        }

        @Override
        public void connected(Channel channel) throws RemotingException {
            invoke(channel, Constants.ON_CONNECT_KEY);
        }

        @Override
        public void disconnected(Channel channel) throws RemotingException {
            if (logger.isInfoEnabled()) {
                logger.info("disconected from " + channel.getRemoteAddress() + ",url:" + channel.getUrl());
            }
            invoke(channel, Constants.ON_DISCONNECT_KEY);
        }

        private void invoke(Channel channel, String methodKey) {
            Invocation invocation = createInvocation(channel, channel.getUrl(), methodKey);
            if (invocation != null) {
                try {
                    received(channel, invocation);
                } catch (Throwable t) {
                    logger.warn("Failed to invoke event method " + invocation.getMethodName() + "(), cause: " + t.getMessage(), t);
                }
            }
        }

        private Invocation createInvocation(Channel channel, URL url, String methodKey) {
            String method = url.getParameter(methodKey);
            if (method == null || method.length() == 0) {
                return null;
            }
            RpcInvocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
            invocation.setAttachment(Constants.PATH_KEY, url.getPath());
            invocation.setAttachment(Constants.GROUP_KEY, url.getParameter(Constants.GROUP_KEY));
            invocation.setAttachment(Constants.INTERFACE_KEY, url.getParameter(Constants.INTERFACE_KEY));
            invocation.setAttachment(Constants.VERSION_KEY, url.getParameter(Constants.VERSION_KEY));
            if (url.getParameter(Constants.STUB_EVENT_KEY, false)) {
                invocation.setAttachment(Constants.STUB_EVENT_KEY, Boolean.TRUE.toString());
            }
            return invocation;
        }
    };

從上可以看出在該handler中，定義了與客戶端連接成功／斷開連接／接受到客戶端消息／相應消息，以及創造Invocation的方法。其中的getInvoker(Channel channel, Invocation inv)方法簡碼如下：

        String serviceKey = serviceKey(port, path, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY));
        DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
        return exporter.getInvoker();

這不就是我們剛剛放置到exporterMap中的DubboExporter，而其中的invoker不就是我們的“filter的invokerdelegete對象”。

 

使用Exchangers.bind(providerUrl, ExchangeHandlerAdapter對象)創建ExchangeServer

    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");
        }
        url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
        return getExchanger(url).bind(url, handler);
    }

    public static Exchanger getExchanger(URL url) {
        String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);//header
        return getExchanger(type);
    }

    public static Exchanger getExchanger(String type) {
        return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
    }

getExchanger(URL url)返回一個HeaderExchanger實例。所以ExchangeServer的創建交由HeaderExchanger來實現。

 

HeaderExchanger.bind(providerUrl, ExchangeHandlerAdapter對象) 

    /**
     * 1 對handler進行三次包裝：首先將ExchangeHandlerAdapter賦給HeaderExchangeHandler中的ExchangeHandler handler屬性；然后將創建出來的HeaderExchangeHandler賦給DecodeHandler的父類AbstractChannelHandlerDelegate的ChannelHandler handler屬性
     */
    public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    }

說明：

• 這里首先對傳入的ExchangeHandlerAdapter進行了兩次包裝，最終得到DecodeHandler實例；
• 之后，使用Transporters.bind(providerUrl, DecodeHandler對象)創建了一個NettyServer；
• 最后使用HeaderExchangeServer包裝了上邊的NettyServer，並啟動了心跳計數器。
  • HeaderExchangeServer實例也是最終返回的ExchangeServer實例，將最終被存儲在Map<String, ExchangeServer> serverMap：{ "10.10.10.10:20880"<->HeaderExchangeServer實例 }

包裝ExchangeHandlerAdapter，獲取DecodeHandler實例。代碼比較簡單，不列出來了。

最終獲取到的DecodeHandler實例的層級關系：

DecodeHandler實例
-->HeaderExchangeHandler實例
   -->ExchangeHandlerAdapter實例

 

使用Transporters.bind(providerUrl, DecodeHandler對象)創建了一個NettyServer

Transporters.bind(providerUrl, DecodeHandler對象)

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

    public static Transporter getTransporter() {
        return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
    }

 

Transporter$Adaptive.bind(providerUrl, DecodeHandler對象)

    public com.alibaba.dubbo.remoting.Server bind(com.alibaba.dubbo.common.URL arg0, com.alibaba.dubbo.remoting.ChannelHandler arg1) throws com.alibaba.dubbo.remoting.RemotingException {
        if (arg0 == null)
            throw new IllegalArgumentException("url == null");
        com.alibaba.dubbo.common.URL url = arg0;
        String extName = url.getParameter("server", url.getParameter("transporter", "netty"));//netty
        if(extName == null)
            throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.remoting.Transporter) name from url(" + url.toString() + ") use keys([server, transporter])");
        com.alibaba.dubbo.remoting.Transporter extension = (com.alibaba.dubbo.remoting.Transporter)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.remoting.Transporter.class).getExtension(extName);
        return extension.bind(arg0, arg1);
    }

 

最后NettyServer的創建由NettyTransporter來創建。

NettyTransporter.bind(providerUrl, DecodeHandler對象)

public class NettyTransporter implements Transporter {
    public static final String NAME = "netty";

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

    public Client connect(URL url, ChannelHandler listener) throws RemotingException {
        return new NettyClient(url, listener);
    }
}

 

new NettyServer(providerUrl, DecodeHandler對象)

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

這里首先為providerUrl添加參數：threadname=DubboServerHandler-10.10.10.10:20880（ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)）；

之后，使用ChannelHandlers.wrap(DecodeHandler對象, providerUrl)對DecodeHandler對象進行了三層包裝，最終得到MultiMessageHandler實例；

最后調用父類的構造器初始化NettyServer的各個屬性，最后啟動netty。

先看一下

ChannelHandlers.wrap(DecodeHandler對象, providerUrl)

    /**
     * 這里又是層層包裹：
     * MultiMessageHandler
     * --HeartbeatHandler
     *   --AllChannelHandler
     *     --DecodeHandler
     *       --HeaderExchangeHandler
     *         --ExchangeHandlerAdapter
     * @param handler
     * @param url
     * @return
     */
    protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
        return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
                .getAdaptiveExtension().dispatch(handler, url)));
    }

ExtensionLoader.getExtensionLoader(Dispatcher.class).getAdaptiveExtension()獲取到一個Dispatcher$Adaptive適配類。

Dispatcher$Adaptive.dispatch(DecodeHandler對象, providerUrl)

    public com.alibaba.dubbo.remoting.ChannelHandler dispatch(com.alibaba.dubbo.remoting.ChannelHandler arg0, com.alibaba.dubbo.common.URL arg1) {
        if (arg1 == null)
            throw new IllegalArgumentException("url == null");
        com.alibaba.dubbo.common.URL url = arg1;
        String extName = url.getParameter("dispatcher", url.getParameter("dispather", url.getParameter("channel.handler", "all")));//all
        if(extName == null)
            throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.remoting.Dispatcher) name from url(" + url.toString() + ") use keys([dispatcher, dispather, channel.handler])");
        com.alibaba.dubbo.remoting.Dispatcher extension = (com.alibaba.dubbo.remoting.Dispatcher)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.remoting.Dispatcher.class).getExtension(extName);
        return extension.dispatch(arg0, arg1);
    }

這里獲取到AllDispatcher，Dispatcher決定了dubbo的線程模型，指定了哪些做什么，哪些線程做什么。講到dubbo通信的時候再寫。

AllDispatcher.dispatch(DecodeHandler對象, providerUrl)

    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return new AllChannelHandler(handler, url);
    }

new AllChannelHandler(DecodeHandler對象, providerUrl)

    public AllChannelHandler(ChannelHandler handler, URL url) {
        super(handler, url);
    }

來看其父類的WrappedChannelHandler的構造器：

WrappedChannelHandler(DecodeHandler對象, providerUrl)

    protected static final ExecutorService SHARED_EXECUTOR = Executors.newCachedThreadPool(new NamedThreadFactory("DubboSharedHandler", true));
    protected final ExecutorService executor;
    protected final ChannelHandler handler;
    protected final URL url;

    public WrappedChannelHandler(ChannelHandler handler, URL url) {
        this.handler = handler;
        this.url = url;
        executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);

        String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
        if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) {
            componentKey = Constants.CONSUMER_SIDE;
        }
        DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
        dataStore.put(componentKey, Integer.toString(url.getPort()), executor);//{"java.util.concurrent.ExecutorService":{"20880":executor}}
    }

首先創建了一個共享線程池：SHARED_EXECUTOR；

之后為handler／url／executor賦值，其中executor是一個200個線程數的fixed線程池（隊列為0，即同步隊列）；

    public Executor getExecutor(URL url) {
        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);//默認為dubbo，但是我們這里是DubboServerHandler-10.10.10.10:20880（就是之前設置到url上的threadname）
        int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);//200
        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);//0
        return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
                queues == 0 ? new SynchronousQueue<Runnable>() :
                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                                : new LinkedBlockingQueue<Runnable>(queues)),
                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }

之后獲取了一個數據存儲器：SimpleDataStore；

最后將{"java.util.concurrent.ExecutorService":{"20880": executor}}數據存儲在SimpleDataStore的ConcurrentMap<String, ConcurrentMap<String, Object>> data數據結構中。也就是說：每一個端口，有一個線程池。

注意：為什么SimpleDataSource可以做緩存來使用？

    public T getExtension(String name) {
        if (name == null || name.length() == 0)
            throw new IllegalArgumentException("Extension name == null");
        if ("true".equals(name)) {
            return getDefaultExtension();
        }
        Holder<Object> holder = cachedInstances.get(name);
        if (holder == null) {
            cachedInstances.putIfAbsent(name, new Holder<Object>());
            holder = cachedInstances.get(name);
        }
        Object instance = holder.get();
        if (instance == null) {
            synchronized (holder) {
                instance = holder.get();
                if (instance == null) {
                    instance = createExtension(name);
                    holder.set(instance);
                }
            }
        }
        return (T) instance;
    }

其實，就是這樣SimpleDataStore實例會存儲在cachedInstances緩存中，下一次不會再創建，而是直接獲取該緩存。

 

這樣之后，一個AllChannelHandler實例就完成了，該實例屬性如下：

• WrappedChannelHandler.url：dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&channel.readonly.sent=true&codec=dubbo&dubbo=2.0.0&generic=false&heartbeat=60000&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=1287&side=provider&threadname=DubboServerHandler-10.10.10.10:20880&timestamp=1507116859919
• WrappedChannelHandler.handler：DecodeHandler對象
• WrappedChannelHandler.executor：FixedThreadPool實例

當然還有一個類變量WrappedChannelHandler.SHARED_EXECUTOR=CachedThreadPool實例。

 

之后AllChannelHandler實例會被HeartbeatHandler進行包裹，之后HeartbeatHandler實例又會被MultiMessageHandler所包裹，最后得到的MultiMessageHandler實例的層級結構如下：

MultiMessageHandler
-->handler: HeartbeatHandler
   -->handler: AllChannelHandler
         -->url: providerUrl
         -->executor: FixedExecutor
         -->handler: DecodeHandler
            -->handler: HeaderExchangeHandler
               -->handler: ExchangeHandlerAdapter

 

MultiMessageHandler實例創建出來之后，NettyServer就開始調用其各個父類進行屬性的初始化了。首先來看一下NettyServer的父類層級圖：

AbstractServer：

    protected static final String SERVER_THREAD_POOL_NAME = "DubboServerHandler";
    ExecutorService executor;
    private InetSocketAddress localAddress;
    private InetSocketAddress bindAddress;
    private int accepts;
    private int idleTimeout = 600;

    public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
        super(url, handler);
        localAddress = getUrl().toInetSocketAddress();
        String host = url.getParameter(Constants.ANYHOST_KEY, false)
                || NetUtils.isInvalidLocalHost(getUrl().getHost())
                ? NetUtils.ANYHOST : getUrl().getHost();
        bindAddress = new InetSocketAddress(host, getUrl().getPort());
        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()));
    }

首先調用父類初始化屬性，之后啟動服務。

AbstractEndpoint：

    private Codec2 codec;
    private int timeout;
    private int connectTimeout;

    public AbstractEndpoint(URL url, ChannelHandler handler) {
        super(url, handler);
        this.codec = getChannelCodec(url);
        this.timeout = url.getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);//1000
        this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT);//3000
    }

AbstractPeer：

    private final ChannelHandler handler;
    private volatile URL url;

    public AbstractPeer(URL url, ChannelHandler handler) {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
        if (handler == null) {
            throw new IllegalArgumentException("handler == null");
        }
        this.url = url;
        this.handler = handler;
    }

來看一下最后初始化好的NettyServer實例：

• url：providerUrl（dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&channel.readonly.sent=true&codec=dubbo&dubbo=2.0.0&generic=false&heartbeat=60000&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=1287&side=provider&timestamp=1507116859919）
• handler：MultiMessageHandler實例
• codec：DubboCountCodec實例
• timeout：1000
• connectTimeout：3000
• idleTime：600*1000
• localAddress：10.10.10.10:20880
• bindAddress：0.0.0.0:20880
• accepts：0
• executor：null（此時的executor還沒實例話，要等netty服務起來之后才會從緩存中獲取之前存儲在SimpleDataStore緩存中的那個200個線程數的FixedThreadPool實例）

 

之后，就要啟動netty服務了。

    /**
     * 啟動netty服務，監聽客戶端連接
     */
    @Override
    protected void doOpen() throws Throwable {
        NettyHelper.setNettyLoggerFactory();
        ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
        ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
        ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
        bootstrap = new ServerBootstrap(channelFactory);

        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
        channels = nettyHandler.getChannels();
        // https://issues.jboss.org/browse/NETTY-365
        // https://issues.jboss.org/browse/NETTY-379
        // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            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());
    }

說明：

• boss線程數默認只有一個；
• worker線程數：Runtime.getRuntime().availableProcessors() + 1，為計算機核數+1；
• 服務端邏輯處理器為NettyHandler：
• 編碼器為：InternalEncoder實例，內部使用NettyServer的DubboCountCodec實例來編碼
• 解碼器為：InternalDecoder實例，內部使用NettyServer的DubboCountCodec實例來解碼

 NettyHandler：

@Sharable
public class NettyHandler extends SimpleChannelHandler {
    private final Map<String, Channel> channels = new ConcurrentHashMap<String, Channel>(); // <ip:port, channel>
    private final URL url;
    private final ChannelHandler handler;

    public NettyHandler(URL url, ChannelHandler handler) {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
        if (handler == null) {
            throw new IllegalArgumentException("handler == null");
        }
        this.url = url;
        this.handler = handler;
    }

    public Map<String, Channel> getChannels() {
        return channels;
    }

    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            if (channel != null) {
                channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()), channel);
            }
            handler.connected(channel);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            channels.remove(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()));
            handler.disconnected(channel);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.received(channel, e.getMessage());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        super.writeRequested(ctx, e);
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.sent(channel, e.getMessage());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.caught(channel, e.getCause());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }
}

說明：

屬性

• handler：就是當前的NettyServer實例
• url：providerUrl
• channels：存放連接到來的channel

監聽連接完成／連接斷開／接收到消息／發送完消息／異常捕捉事件，之后使用NettyServer實例進行相應的處理，NettyServer又會調用MultiMessageHandler實例（該handler屬性位於NettyServer的父類AbstractPeer中）進行處理。

 

在來看編碼器和解碼器：

NettyCodecAdapter(DubboCountCodec實例, providerUrl, 當前的NettyServer實例)

final class NettyCodecAdapter {
    private final ChannelHandler encoder = new InternalEncoder();
    private final ChannelHandler decoder = new InternalDecoder();
    private final Codec2 codec;
    private final URL url;
    private final int bufferSize;
    private final com.alibaba.dubbo.remoting.ChannelHandler handler;

    public NettyCodecAdapter(Codec2 codec, URL url, com.alibaba.dubbo.remoting.ChannelHandler handler) {
        this.codec = codec;
        this.url = url;
        this.handler = handler;
        int b = url.getPositiveParameter(Constants.BUFFER_KEY, Constants.DEFAULT_BUFFER_SIZE);//8*1024
        this.bufferSize = b >= Constants.MIN_BUFFER_SIZE && b <= Constants.MAX_BUFFER_SIZE ? b : Constants.DEFAULT_BUFFER_SIZE;//8*1024
    }

    public ChannelHandler getEncoder() {
        return encoder;
    }

    public ChannelHandler getDecoder() {
        return decoder;
    }

    @Sharable
    private class InternalEncoder extends OneToOneEncoder {
        @Override
        protected Object encode(ChannelHandlerContext ctx, Channel ch, Object msg) throws Exception {
            ...
            codec.encode(channel, buffer, msg);
            ...
        }
    }

    private class InternalDecoder extends SimpleChannelUpstreamHandler {
        @Override
        public void messageReceived(ChannelHandlerContext ctx, MessageEvent event) throws Exception {
           ...
            msg = codec.decode(channel, message);
           ...
        }
        ...
    }
}

只列出核心代碼：可以看到，InternalEncoder實例和InternalDecoder實例內部還是使用NettyServer的DubboCountCodec實例來編解碼的。dubbo的編解碼做的非常好，后續會寫。

 

到此為止，NettyServer就創建成功了。 之后，終於執行到了：

new HeaderExchangeServer(Server NettyServer)。

    private final ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1, new NamedThreadFactory("dubbo-remoting-server-heartbeat",true));
    private final Server server;
    // 心跳定時器
    private ScheduledFuture<?> heatbeatTimer;
    // 心跳超時，毫秒。缺省0，不會執行心跳。
    private int heartbeat;
    private int heartbeatTimeout;
    private AtomicBoolean closed = new AtomicBoolean(false);

    public HeaderExchangeServer(Server server) {
        if (server == null) {
            throw new IllegalArgumentException("server == null");
        }
        this.server = server;
        this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);//60000 在createServer(URL providerUrl)中拼接了heartbeat參數
        this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3);//3*60000
        if (heartbeatTimeout < heartbeat * 2) {
            throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2");
        }
        startHeatbeatTimer();
    }

說明：

• 屬性
  • scheduled：是一個有1個名字為dubbo-remoting-server-heartbeat的后台線程的定時線程池；
  • server：之前創建出來的NettyServer實例；
  • heartbeatTimer：心跳計時器
  • heartbeat：心跳時間，該參數會在HeaderExchangeServer的構造器中進行賦值，60000
  • heartbeatTimeout：心跳超時時間，超過該時間，會進行channel重連，180000
• 啟動心跳計時器

startHeatbeatTimer()

    private void startHeatbeatTimer() {
        stopHeartbeatTimer();
        if (heartbeat > 0) {
            heatbeatTimer = scheduled.scheduleWithFixedDelay(
                    new HeartBeatTask(new HeartBeatTask.ChannelProvider() {
                        public Collection<Channel> getChannels() {
                            return Collections.unmodifiableCollection(HeaderExchangeServer.this.getChannels());
                        }
                    }, heartbeat, heartbeatTimeout),
                    heartbeat,
                    heartbeat,
                    TimeUnit.MILLISECONDS);
        }
    }

    private void stopHeartbeatTimer() {
        try {
            ScheduledFuture<?> timer = heatbeatTimer;
            if (timer != null && !timer.isCancelled()) {
                timer.cancel(true);
            }
        } catch (Throwable t) {
            logger.warn(t.getMessage(), t);
        } finally {
            heatbeatTimer = null;
        }
    }

首先停掉之前的計時器，之后在線程創建開始heartbeat毫秒（60s）后執行第一次HeartBeatTask任務，之后每隔heartbeat毫秒（60s）執行一次HeartBeatTask任務。來看一下HeartBeatTask：

HeartBeatTask

final class HeartBeatTask implements Runnable {
    private ChannelProvider channelProvider;
    private int heartbeat;//60s
    private int heartbeatTimeout;//180s

    HeartBeatTask(ChannelProvider provider, int heartbeat, int heartbeatTimeout) {
        this.channelProvider = provider;
        this.heartbeat = heartbeat;
        this.heartbeatTimeout = heartbeatTimeout;
    }

    public void run() {
        try {
            long now = System.currentTimeMillis();
            for (Channel channel : channelProvider.getChannels()) {
                if (channel.isClosed()) {
                    continue;
                }
                try {
                    Long lastRead = (Long) channel.getAttribute(HeaderExchangeHandler.KEY_READ_TIMESTAMP);//"READ_TIMESTAMP"
                    Long lastWrite = (Long) channel.getAttribute(HeaderExchangeHandler.KEY_WRITE_TIMESTAMP);//"WRITE_TIMESTAMP"
                    //如果最后一次讀和寫在heartbeat時間（60s）內，則最后一次的讀和寫本身可以看作心跳；否則，需要程序發送心跳
                    if ((lastRead != null && now - lastRead > heartbeat)
                            || (lastWrite != null && now - lastWrite > heartbeat)) {
                        Request req = new Request();
                        req.setVersion("2.0.0");
                        req.setTwoWay(true);
                        req.setEvent(Request.HEARTBEAT_EVENT);
                        channel.send(req);
                        if (logger.isDebugEnabled()) {
                            logger.debug("Send heartbeat to remote channel " + channel.getRemoteAddress()
                                    + ", cause: The channel has no data-transmission exceeds a heartbeat period: " + heartbeat + "ms");
                        }
                    }
                    //如果最后一次讀的時間距離現在已經超過heartbeatTimeout了，我們認為channel已經斷了（因為在這個過程中，發送了三次心跳都沒反應），此時channel進行重連
                    if (lastRead != null && now - lastRead > heartbeatTimeout) {
                        logger.warn("Close channel " + channel
                                + ", because heartbeat read idle time out: " + heartbeatTimeout + "ms");
                        if (channel instanceof Client) {
                            try {
                                ((Client) channel).reconnect();
                            } catch (Exception e) {
                                //do nothing
                            }
                        } else {
                            channel.close();
                        }
                    }
                } catch (Throwable t) {
                    logger.warn("Exception when heartbeat to remote channel " + channel.getRemoteAddress(), t);
                }
            }
        } catch (Throwable t) {
            logger.warn("Unhandled exception when heartbeat, cause: " + t.getMessage(), t);
        }
    }

    interface ChannelProvider {
        Collection<Channel> getChannels();
    }
}

說明：

• 屬性
  • channelProvider在startHeatbeatTimer()中創建，並且獲取了當前的HeaderExchangeServer的所有channels
  • heartbeat：60s
  • heartbeatTimeout：180s
• run()
  • 如果最后一次讀和寫的時間距離現在在heartbeat時間（60s）內，則最后一次的讀和寫本身可以看作心跳；否則，發送心跳
  • 如果最后一次讀的時間距離現在已經超過heartbeatTimeout了，認為channel已經斷了（因為在這個過程中，發送了三次心跳都沒反應），此時channel進行重連

到現在一個完整的ExchangeServer就OK了。之后我們將創建出來的ExchangeServer實例存放在DubboProtocol的Map<String, ExchangeServer> serverMap屬性中：

{ "10.10.10.10:20880" : ExchangeServer實例 }

最后，DubboProtocol.export(Invoker<T> invoker)將之前創建的DubboExporter實例返回。

 

2.4  創建RegistryProtocol.ExporterChangeableWrapper來封裝Exporter和originInvoker

exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker)

    private class ExporterChangeableWrapper<T> implements Exporter<T> {
        private final Invoker<T> originInvoker;
        private Exporter<T> exporter;

        public ExporterChangeableWrapper(Exporter<T> exporter, Invoker<T> originInvoker) {
            this.exporter = exporter;
            this.originInvoker = originInvoker;
        }

        public Invoker<T> getOriginInvoker() {
            return originInvoker;
        }

        public Invoker<T> getInvoker() {
            return exporter.getInvoker();
        }

        public void setExporter(Exporter<T> exporter) {
            this.exporter = exporter;
        }

        public void unexport() {
            String key = getCacheKey(this.originInvoker);
            bounds.remove(key);
            exporter.unexport();
        }
    }

ExporterChangeableWrapper類是RegistryProtocol的私有內部類。

最后，將<providerUrl, ExporterChangeableWrapper實例>放入RegistryProtocol的屬性Map<String, ExporterChangeableWrapper<?>> bounds中。

• key：dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=744&side=provider&timestamp=1507176748026
• value：RegistryProtocol$ExporterChangeableWrapper實例
  • originInvoker：即AbstractProxyInvoker實例屬性如下：
    • proxy：DemoServiceImpl實例
    • type：Class<com.alibaba.dubbo.demo.DemoService>
    • url：registry://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&export=dubbo%3A%2F%2F10.10.10.10%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D993%26side%3Dprovider%26timestamp%3D1507100322516&pid=993&registry=zookeeper&timestamp=1507100319830
  • DubboExporter實例
    • key：com.alibaba.dubbo.demo.DemoService:20880
    • invoker："InvokerDelegete的filter對象"
    • exporterMap：{ "com.alibaba.dubbo.demo.DemoService:20880" -> 當前的DubboExporter實例 }

 

到此為止，RegistryProtocol.export(final Invoker<T> originInvoker)的第一行代碼就完成了。

    public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        //export invoker
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        //registry provider
        final Registry registry = getRegistry(originInvoker);
        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
        registry.register(registedProviderUrl);
        // 訂閱override數據
        // FIXME 提供者訂閱時，會影響同一JVM即暴露服務，又引用同一服務的的場景，因為subscribed以服務名為緩存的key，導致訂閱信息覆蓋。
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        //保證每次export都返回一個新的exporter實例
        return new Exporter<T>() {
            public Invoker<T> getInvoker() {
                return exporter.getInvoker();
            }

            public void unexport() {
                try {
                    exporter.unexport();
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    registry.unregister(registedProviderUrl);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    overrideListeners.remove(overrideSubscribeUrl);
                    registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
            }
        };
    }