Dubbo服務合買平臺搭建出售發佈之服務暴露&心跳機制&服務註冊

Dubbo服務發佈

Dubbo合買平臺搭建出售 dsluntan.com Q:3393756370 VX:17061863513服務發佈影響流程的主要包括三個部分，依次是：

1. 服務暴露
2. 心跳
3. 服務註冊

服務暴露是對外提供服務及暴露端口，以便消費端能夠正常調通服務。心跳機制保證服務器端及客戶端正常長鏈接的保持，服務註冊是向註冊中心註冊服務暴露服務的過程。

Dubbo服務暴露

此處只記錄主要代碼部分以便能快速定位到主要的核心代碼：

ServiceConfig.java中代碼

if (registryURLs != null && registryURLs.size() > 0 && url.getParameter("register", true)) { // 循環祖冊中心 URL 數組 registryURLs for (URL registryURL : registryURLs) { // "dynamic" ：服務是否動態註冊，若是設爲false，註冊後將顯示後disable狀態，需人工啓用，而且服務提供者中止時，也不會自動取消冊，需人工禁用。 url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic")); // 得到監控中心 URL URL monitorUrl = loadMonitor(registryURL); if (monitorUrl != null) { // 將監控中心的 URL 做爲 "monitor" 參數添加到服務提供者的 URL 中，而且須要編碼。經過這樣的方式，服務提供者的 URL 中，包含了監控中心的配置。 url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString()); } if (logger.isInfoEnabled()) { logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL); } // 使用 ProxyFactory 建立 Invoker 對象 // 調用 URL#addParameterAndEncoded(key, value) 方法，將服務體用這的 URL 做爲 "export" 參數添加到註冊中心的 URL 中。經過這樣的方式，註冊中心的 URL 中，包含了服務提供者的配置。 // 建立 Invoker 對象。該 Invoker 對象，執行 #invoke(invocation) 方法時，內部會調用 Service 對象( ref )對應的調用方法。 Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); // 使用 Protocol 暴露 Invoker 對象 /** * Protocol$Adaptive => ProtocolFilterWrapper => ProtocolListenerWrapper => RegistryProtocol * => * Protocol$Adaptive => ProtocolFilterWrapper => ProtocolListenerWrapper => DubboProtocol */ Exporter<?> exporter = protocol.export(invoker); // 添加到 `exporters` exporters.add(exporter); } }

循環註冊中心，對每一個註冊中心都執行代碼塊中的執行過程

1.若是url中沒有dynamic 參數，則從registerUrl中取值，並賦予url
dynamic是服務動態註冊的標識，默認爲true，若是設置爲false,則服務註冊後顯示disable狀態，需人工啓動

2.加載註冊中心對應的監控中心配置

3.若是註冊中心不爲空則設置url的 monitor參數

4.Invoker proxyFactory.getInvoker  proxyFactory 默認爲JavassistProxyFactory對象，這段代碼爲建立 ref 服務對象的代理對象。
proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); 獲取ref的代理對象並在registryURL 中添加export屬性，代理對象中屬性參數以下

5.protocol.export(invoker) 爲暴露服務的核心實現部分,協議的調用鏈以下：

 /**
 * Protocol$Adaptive => ProtocolFilterWrapper => ProtocolListenerWrapper => RegistryProtocol
 * =>
 * Protocol$Adaptive => ProtocolFilterWrapper => ProtocolListenerWrapper => DubboProtocol
 */

其中DubboProtocol 實現了服務暴露及心跳檢測功能 
RegistryProtocol 調用了DubboProtocol及註冊服務

接下來通過兩個擴展類(包裝器) ProtocolFilterWrapper和ProtocolListenerWrapper 進入RegistryProtocol 核心代碼以下：

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); // 得到服務提供者 URL 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); } } }; }

 

/** * 暴露服務。 * * 此處的 Local 指的是，本地啓動服務，可是不包括向註冊中心註冊服務的意思。 * @param originInvoker * @param <T> * @return */ @SuppressWarnings("unchecked") private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) { // 得到在 `bounds` 中的緩存 Key //dubbo://192.168.20.218:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&default.accepts=1000&default.threadpool=fixed&default.threads=100&default.timeout=5000&dubbo=2.0.0&generic=false& // interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&owner=uce&pid=1760&side=provider&timestamp=1530150456618 String key = getCacheKey(originInvoker); ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key); if (exporter == null) { synchronized (bounds) { exporter = (ExporterChangeableWrapper<T>) bounds.get(key); // 未暴露過，進行暴露服務 if (exporter == null) { // InvokerDelegete 實現 com.alibaba.dubbo.rpc.protocol.InvokerWrapper 類，主要增長了 #getInvoker() 方法，得到真實的，非 InvokerDelegete 的 Invoker 對象。 // 由於，可能會存在 InvokerDelegete.invoker 也是 InvokerDelegete 類型的狀況。 getProviderUrl 同上 key = getCacheKey final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker)); // 暴露服務，建立 Exporter 對象 Exporter<T> export = (Exporter<T>) protocol.export(invokerDelegete); // 使用 建立的Exporter對象 + originInvoker ，建立 ExporterChangeableWrapper 對象 exporter = new ExporterChangeableWrapper<T>(export, originInvoker); bounds.put(key, exporter); } } } return exporter; }

1.代用同步鎖+double-check的方式來保證一樣的服務不重複暴露。

2.new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
InvokerDelegete 實現 com.alibaba.dubbo.rpc.protocol.InvokerWrapper（invoke） 類，主要增長了 #getInvoker() 方法，得到真實的，非 InvokerDelegete 的 Invoker 對象。

3.調用protocol.export接口 通過ProtocolFilterWrapper.invoker方法 創過濾器鏈再暴露服務：

protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));

/** * 構建過濾器鏈 * @param invoker injvm://127.0.0.1/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&default.accepts=1000&default.threadpool=fixed&default.threads=100&default.timeout=5000&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&owner=uce&pid=9932&side=provider&timestamp=1527930395583 * @param key service.filter 該參數用於得到 ServiceConfig 或 ReferenceConfig 配置的自定義過濾器 * 以 ServiceConfig 舉例子，例如 url = injvm://127.0.0.1/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=192.168.3.17&bind.port=20880&default.delay=-1&default.retries=0&default.service.filter=demo&delay=-1&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=81844&qos.port=22222&service.filter=demo&side=provider&timestamp=1520682156043 中， * service.filter=demo，這是筆者配置自定義的 DemoFilter 過濾器。 * <dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService" filter="demo" /> * @param group provider 屬性，分組 * 在暴露服務時，group = provider 。 * 在引用服務時，group = consumer 。 * @param <T> * @return */ 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); /* EchoFilter ClassLoaderFilter GenericFilter ContextFilter TraceFilter TimeoutFilter MonitorFilter ExceptionFilter DemoFilter 【自定義】*/ //倒序循環 Filter ，建立帶 Filter 鏈的 Invoker 對象。由於是經過嵌套聲明匿名類循環調用的方式，因此要倒序。能夠手工模擬下這個過程。經過這樣的方式，實際過濾的順序，仍是咱們上面看到的正序 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>() { @Override public Class<T> getInterface() { return invoker.getInterface(); } @Override public URL getUrl() { return invoker.getUrl(); } @Override public boolean isAvailable() { return invoker.isAvailable(); } @Override public Result invoke(Invocation invocation) throws RpcException { return filter.invoke(next, invocation); } @Override public void destroy() { invoker.destroy(); } @Override public String toString() { return invoker.toString(); } }; } } return last; }

List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
獲取Active的屬於指定組的過過濾器列表
參考文章：https://my.oschina.net/LucasZhu/blog/1835048

接下來執行DubboProrocol進行服務暴露的過程。

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { URL url = invoker.getUrl(); // export service. String key = serviceKey(url); // 建立 DubboExporter 對象，並添加到 `exporterMap` 。 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; }

1.獲取invoker的 URL信息
2.獲取key信息 爲URL中interface與暴露端口的拼裝字符串：com.alibaba.dubbo.demo.DemoService:20880
3.建立DubboExporter對象 而且入參爲exporterMap
4.將exporter對象添加到exporterMap中
 

/** * 啓動服務器 * * @param url URL */ private void openServer(URL url) { // find server. String key = url.getAddress(); //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); } } }

調用createServer()方法 並存入DubboProtocol的serverMap中
 

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); // 校驗 Server 的 Dubbo SPI 拓展是否存在 if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) throw new RpcException("Unsupported server type: " + str + ", url: " + url); // 設置codec爲 `"Dubbo"` 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; }

1.默認開啓server 關閉時發送readonly事件：channel.readonly.sent : true
2.默認開啓 heartbeat 
3.獲取服務暴露的 server 傳輸 ， 默認爲netty
4.設置編碼器爲Dubbo也就是 DubboCountCodec
5.Exchangers#bind(url, requestHandler) 啓動服務器，requestHandler結構以下

具體實現代碼以下：

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

Exchangeers.bind(URL url, ExchangeHandler handler)

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); return getExchanger(type); } public static Exchanger getExchanger(String type) { return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type); }

接口做用是設置exchanger params爲header 而且獲取Exchanger.class的header擴展接口HeaderExchanger， 並調用bind方法：

public class HeaderExchanger implements Exchanger { public static final String NAME = "header"; public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true); } public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); } }

先將 DubboProtocol入參 傳過來的ExchangeHandler對象ExchangeHandlerAdapter() 進行包裝組成handler鏈：最後返回ChannelHandler對象，接下來調用：Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))
Server Transporters.bind(URL url, ChannelHandler... handlers)
Transpoter$Adaptive.bind()

數據透傳 NettyTransporter.java
Server NettyTransporter.bind(URL url, ChannelHandler listener)

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

做用是：

返回一個NettyServer實例：

 

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

ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)) 只用是生成獲取ThreadName的名稱 爲URL添加threadname的param
ChannelHandlers.wrap(ChannelHandler handler, URL url)  代碼以下：

public class ChannelHandlers { private static ChannelHandlers INSTANCE = new ChannelHandlers(); protected ChannelHandlers() { } public static ChannelHandler wrap(ChannelHandler handler, URL url) { return ChannelHandlers.getInstance().wrapInternal(handler, url); } protected static ChannelHandlers getInstance() { return INSTANCE; } static void setTestingChannelHandlers(ChannelHandlers instance) { INSTANCE = instance; } 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().dispatch(handler, url)：

獲取到AllDispatcher分發器進行透傳：

public class AllDispatcher implements Dispatcher { public static final String NAME = "all"; public ChannelHandler dispatch(ChannelHandler handler, URL url) { return new AllChannelHandler(handler, url); } }

結構如圖所示：

調用WrappedChannelHandler的構造方法：

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

這段代碼的功能爲：

1.將 以前頭創的DecoderHandler對象再進包裝 包裝爲AllChannelHandler
2.生成線程池對象Executor對象
3.獲取默認的DataStore對象，並將線程池對象放入DataStore 中 key爲 : java.util.concurrent.ExecutorService 字符串和服務暴露的端口 值爲線程池對象

return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
                .getAdaptiveExtension().dispatch(handler, url)));
接下來將返回的AllChannelHandler對象用HeartbeatHandler 和 MultiMessageHandler 進行包裝處理並返回ChannelHandler.wrap() 的上一端。

NettyTransporter.bind(URL url, ChannelHandler listener) -> new NettyServer(URL url, ChannelHandler handler)
->  super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
接下來是建立NettyServer對象的最後一步：

NettyServer ==>
public NettyServer(URL url, ChannelHandler handler) throws RemotingException { super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME))); } AbstractServer==> 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 ==> public AbstractEndpoint(URL url, ChannelHandler handler) { super(url, handler); this.codec = getChannelCodec(url); this.timeout = url.getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT); this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT); } AbstractPeer==> 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; }

調用棧如上所示：
由於以前設置了codec爲dubbo 因此返回DubboCountCodec實例
獲取超時時間timeout ,和連接的超時時間connectTimeout
localAddress爲本地IP:PORT  port爲服務暴露的端口
host 爲0.0.0.0
bindAddress爲 host:port port爲服務暴露的端口
this.accept 爲默認獲取最大鏈接數
idleTimeout爲 url中 idle.timeout
核心代碼：doOpen()

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

1.首先進行Netty的日誌配置
接下來先生成 NettyCodecAdapter 入參爲以前生成的codec , URL信息(主要用到buffer屬性配置Netty緩衝區)及 this (Handler) 對象
接下來就是設置Netty的Encoder Decoder 來進行數據的編碼與解碼 其會調用 this的handler鏈來進行數據處理。Dubbo2.5.6採用的是Netty3來進行通信的，此處就不進行贅述。

AbstractServer 接下來獲取到從DataStore對象中獲取以前緩存的線程池 ，設置 NettyServer的 executor屬性。

自此，Dubbo服務暴露的代碼解析完畢，NettyServer的類結構圖以下：

心跳服務

Dubbo provider的心跳服務是 HeaderExchanger bind代碼執行的最後一步：參數是上面生成的Server對象 (NettyServer)。

public HeaderExchangeServer(Server server) { if (server == null) { throw new IllegalArgumentException("server == null"); } this.server = server; this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0); this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3); if (heartbeatTimeout < heartbeat * 2) { throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2"); } startHeatbeatTimer(); }

1.初始化 server信息
2.獲取server URL中heartbeat信息 及心跳超時信息，默認爲heartbeat的三倍
3.執行心跳代碼 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); } }

1.中止定時任務——首先中止定時器中全部任務，置空 beatbeatTimer；
2.從新設置定時器 ， 循環檢測

接下來在DubboProtocol的openServer(URL) 方法中將建立的ExchangeServer對象放入 DubboProtocol的 serverMap 集合對象中 
key爲服務的ip:port 如 192.168.20.218:20880
value爲以前建立的ExchangeServer對象

DubboProtocol export方法到此執行完畢，最終返回的是 DubboExporter對象包裝了入參的invoker對象，serviceKey信息，及服務暴露的 exporterMap對象。

服務註冊

咱們接着來看RegistryProtocol 接下來的執行代碼：

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { // 暴露服務 //export invoker final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker); //registry provider 添加定時任務 ping request response final Registry registry = getRegistry(originInvoker); // 得到服務提供者 URL 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); } } }; }

1.ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) 爲暴露服務的執行過程，上面流程已經走過。
返回的數據格式以下：

2.根據originInvoker中註冊中心信息獲取對應的Registry對象,由於這裏是zookeeper協議，因此爲ZookeeperRegistry對象
3.從註冊中心的URL中得到 export 參數對應的值，即服務提供者的URL.
4.registry.register(registedProviderUrl); 用以前建立的註冊中心對象註冊服務
5.
 

// TODO 

 

上面提到 Registry getRegistry(final Invoker<?> originInvoker) 是根據invoker的地址獲取registry實例代碼以下：

private Registry getRegistry(final Invoker<?> originInvoker) { // registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&export=dubbo%3A%2F%2F192.168.20.218%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26default.accepts%3D1000%26default.threadpool%3Dfixed%26default.threads%3D100%26default.timeout%3D5000%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26owner%3Duce%26pid%3D12028%26side%3Dprovider%26timestamp%3D1531912729429&owner=uce&pid=12028&registry=zookeeper&timestamp=1531912729343 URL registryUrl = originInvoker.getUrl(); if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) { String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY); registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY); } // zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&export=dubbo%3A%2F%2F192.168.20.218%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26default.accepts%3D1000%26default.threadpool%3Dfixed%26default.threads%3D100%26default.timeout%3D5000%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26owner%3Duce%26pid%3D12028%26side%3Dprovider%26timestamp%3D1531912729429&owner=uce&pid=12028&timestamp=1531912729343 return registryFactory.getRegistry(registryUrl); }

上面代碼的意思是：
1.獲取originalInvoker中的URL信息 (註冊中心的配置信息)
2.將URL中信息中Param中registry參數獲取到，並替換URL中的protocol屬性，並刪除Param中的registry信息，上面代碼中的註釋爲執行前和執行後的的結果。
3.獲取protocol 爲 zookeeper對應的RegistryFactory接口的擴展對象 ZookeeperRegistryFactory 並執行getRegistry 方法：

ZookeeperRegistryFactory的繼承結構和對應類中屬性以下圖所示：
其中REGISTRIES = new ConcurrentHashMap<String, Registry>(); 表明註冊中心的配置，其中能夠有多個註冊中心配置

AbstractRegistryFactory.getRegistry執行代碼以下：

public Registry getRegistry(URL url) { url = url.setPath(RegistryService.class.getName()) .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName()) .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY); String key = url.toServiceString(); // zookeeper://192.168.1.157:2181/com.alibaba.dubbo.registry.RegistryService // 鎖定註冊中心獲取過程，保證註冊中心單一實例 LOCK.lock(); try { Registry registry = REGISTRIES.get(key); if (registry != null) { return registry; } registry = createRegistry(url); if (registry == null) { throw new IllegalStateException("Can not create registry " + url); } REGISTRIES.put(key, registry); return registry; } finally { // 釋放鎖 LOCK.unlock(); } }

1.設置Path屬性，添加interface參數信息，及移除export 和 refer 參數信息。執行結果以下：
zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&interface=com.alibaba.dubbo.registry.RegistryService&owner=uce&pid=12028&timestamp=1531912729343
2.獲取url對應的serviceString信息：zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService，因爲我使用的是本地的zookeeper 因此IP爲 127.0.0.1
3.順序地建立註冊中心：Registry ZookeeperRegistryFactory.createRegistry(URL url);

public Registry createRegistry(URL url) { return new ZookeeperRegistry(url, zookeeperTransporter); } // 構造ZookeeperRegistry的調用鏈以下所示 public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) { super(url); if (url.isAnyHost()) { throw new IllegalStateException("registry address == null"); } String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT); if (!group.startsWith(Constants.PATH_SEPARATOR)) { group = Constants.PATH_SEPARATOR + group; } this.root = group; zkClient = zookeeperTransporter.connect(url); zkClient.addStateListener(new StateListener() { public void stateChanged(int state) { if (state == RECONNECTED) { try { recover(); } catch (Exception e) { logger.error(e.getMessage(), e); } } } }); } public FailbackRegistry(URL url) { super(url); int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD); this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() { public void run() { // 檢測並鏈接註冊中心 try { retry(); } catch (Throwable t) { // 防護性容錯 logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t); } } }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS); } public AbstractRegistry(URL url) { setUrl(url); // 啓動文件保存定時器 syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false); String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getHost() + ".cache"); File file = null; if (ConfigUtils.isNotEmpty(filename)) { file = new File(filename); if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) { if (!file.getParentFile().mkdirs()) { throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!"); } } } this.file = file; loadProperties(); notify(url.getBackupUrls()); }

ZookeeperRegistry 的類繼承結構圖如圖所示：

ZooKeeperRegistry.FailbackRegistry.AbstractRegistry中
1.setUrl設置url屬性信息
2.是否啓用文件的異步保存
3.註冊中心對應的本地文件保存的位置信息：如C:\Users\Administrator/.dubbo/dubbo-registry-127.0.0.1.cache
4.給file賦值 而且加載文件信息到properties屬性中
5.notify(url.getBackupUrls) 這段代碼不知道什麼意思。

ZooKeeperRegistry.FailbackRegistry中
1.獲取定時任務的時間間隔。
2.開啓定時任務定時檢測失敗的註冊，並從新註冊。

ZooKeeperRegistry 中
1.獲取註冊中心的group參數 ，默認爲/dubbo , 並未root賦予group值
2.zkClient = zookeeperTransporter.connect(url); 連接zookeeper信息並添加狀態監聽事件，具體再更文詳述吧，代碼以下：

public ZkclientZookeeperClient(URL url) { super(url); client = new ZkClient(url.getBackupAddress()); client.subscribeStateChanges(new IZkStateListener() { @Override public void handleStateChanged(KeeperState state) throws Exception { ZkclientZookeeperClient.this.state = state; if (state == KeeperState.Disconnected) { stateChanged(StateListener.DISCONNECTED); } else if (state == KeeperState.SyncConnected) { stateChanged(StateListener.CONNECTED); } } @Override public void handleNewSession() throws Exception { stateChanged(StateListener.RECONNECTED); } }); }

3.添加劇連狀態的狀態監聽事件 調用 recover()方法。
至此 ZookeeperRegistry建立完畢。

ZookeeperRegistryFactory中最後將registry放入 ZookeeperRegistryFactory.REGISTRIES中 
key 爲zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService
value 爲以前建立的ZookeeperRegistry對象。

接着返回RegistryProtocol 的export方法 ，
1.上面說到了調用doLocalExport(originInvoker);進行服務暴露的過程及調用getRegistry(originInvoker)方法經過ZookeeperRegistryFactory 工廠生成 ZookeeperRegistry 方法，而後加入到工廠REGISTRIES 緩存中，並返回ZookeeperRegistry 實例的過程。

2.接下來RegistryProtocol 的export方法中調用 final URL registedProviderUrl = getRegistedProviderUrl(originInvoker); 獲取服務提供者的URL信息 ， 它是從註冊中心的URL中得到export參數對應的值轉換的URL信息。（去除掉不須要在註冊中心上看到的字段）

3.接下來調用registry.register(registedProviderUrl); 進行服務的註冊將暴露的服務信息註冊到註冊中心，而且將已經註冊的服務URL緩存到ZookeeperRegistry.registered 已註冊服務的緩存中。

FailbackRegistry.register
/** * 進行服務註冊邏輯的實現 */ @Override public void register(URL url) { if (destroyed.get()){ return; } // 調用AbstractRegistry.register進行服務對應URL的緩存 super.register(url); failedRegistered.remove(url); failedUnregistered.remove(url); try { // 向服務器端發送註冊請求，將服務註冊到註冊中心，可使用各個註冊協議(註冊中心)的實現 此處使用zookeeper ZookeeperRegistry.doRegister doRegister(url); } catch (Exception e) { Throwable t = e; // 若是開啓了啓動時檢測，則直接拋出異常 boolean check = getUrl().getParameter(Constants.CHECK_KEY, true) && url.getParameter(Constants.CHECK_KEY, true) && !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol()); boolean skipFailback = t instanceof SkipFailbackWrapperException; if (check || skipFailback) { if (skipFailback) { t = t.getCause(); } throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t); } else { logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t); } // 將失敗的註冊請求記錄到失敗列表，定時重試 failedRegistered.add(url); } } AbstractRegistry.register public void register(URL url) { if (url == null) { throw new IllegalArgumentException("register url == null"); } if (logger.isInfoEnabled()) { logger.info("Register: " + url); } // 緩存已經註冊的服務 registered.add(url); } ZookeeperRegistry.doRegister protected void doRegister(URL url) { try { // 此處爲具體服務暴露的代碼 toUrlPath 根據URL生成寫入zk的路徑信息 zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true)); } catch (Throwable e) { throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e); } }

4.由registryProviderUrl獲取overrideSubscribeUrl 再構建OverrideListener