dubbo源码解析（二）

大家好，我是烤鸭：

dubbo 源码解析：

1.服务导出

介绍:
Dubbo 服务导出过程始于 Spring 容器发布刷新事件，Dubbo 在接收到事件后，会立即执行服务导出逻辑。整个逻辑大致可分为三个部分，第一部分是前置工作，主要用于检查参数，组装 URL。第二部分是导出服务，包含导出服务到本地 (JVM)，和导出服务到远程两个过程。第三部分是向注册中心注册服务，用于服务发现。

源码：

com.alibaba.dubbo.config.spring.ServiceBean

public void onApplicationEvent(ApplicationEvent event) {if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {//如果不延迟,导出if (isDelay() && ! isExported() && ! isUnexported()) {if (logger.isInfoEnabled()) {logger.info("The service ready on spring started. service: " + getInterface());}export();}}
}

服务导出，查看 export 和 isDelay,需要导出或者延迟导出。保留主要代码。

com.alibaba.dubbo.config.ServiceConfig

public synchronized void export() {//如果延迟，另起一条线程，睡眠延迟时间再导出if (delay != null && delay > 0) {Thread thread = new Thread(new Runnable() {public void run() {try {Thread.sleep(delay);} catch (Throwable e) {}doExport();}});thread.setDaemon(true);thread.setName("DelayExportServiceThread");thread.start();} else {doExport();}
}

com.alibaba.dubbo.config.ServiceConfig

doExport 方法较长，就不贴源码了。主要是一堆校验，为的是后续的服务导出。

简单总结：

1. 检测 <dubbo:service> 标签的 interface 属性合法性，不合法则抛出异常
2. 检测 ProviderConfig、ApplicationConfig 等核心配置类对象是否为空，若为空，则尝试从其他配置类对象中获取相应的实例。
3. 检测并处理泛化服务和普通服务类
4. 检测本地存根配置，并进行相应的处理
5. 对 ApplicationConfig、RegistryConfig 等配置类进行检测，为空则尝试创建，若无法创建则抛出异常

多协议多注册中心导出服务 ，这个也不贴源码了。

通过 loadRegistries 加载注册中心链接，然后再遍历 ProtocolConfig 集合导出每个服务。并在导出服务的过程中，将服务注册到注册中心。
com.alibaba.dubbo.config.ServiceConfig.loadRegistries

1. 检测是否存在注册中心配置类，不存在则抛出异常
2. 构建参数映射集合，也就是 map
3. 构建注册中心链接列表
4. 遍历链接列表，并根据条件决定是否将其添加到 registryList 中

组装 URL：
URL 是 Dubbo 配置的载体，通过 URL 可让 Dubbo 的各种配置在各个模块之间传递。
com.alibaba.dubbo.config.ServiceConfig.doExportUrlsFor1Protocol 前半部分
这段代码用于检测 <dubbo:method> 标签中的配置信息，并将相关配置添加到 map 中。

// 获取 ArgumentConfig 列表
for (遍历 ArgumentConfig 列表) {if (type 不为 null，也不为空串) {    // 分支11. 通过反射获取 interfaceClass 的方法列表for (遍历方法列表) {1. 比对方法名，查找目标方法2. 通过反射获取目标方法的参数类型数组 argtypesif (index != -1) {    // 分支21. 从 argtypes 数组中获取下标 index 处的元素 argType2. 检测 argType 的名称与 ArgumentConfig 中的 type 属性是否一致3. 添加 ArgumentConfig 字段信息到 map 中，或抛出异常} else {    // 分支31. 遍历参数类型数组 argtypes，查找 argument.type 类型的参数2. 添加 ArgumentConfig 字段信息到 map 中}}} else if (index != -1) {    // 分支41. 添加 ArgumentConfig 字段信息到 map 中}
}

导出 Dubbo 服务：
服务导出分为导出到本地 (JVM)，和导出到远程。
com.alibaba.dubbo.config.ServiceConfig.doExportUrlsFor1Protocol 后半部分
根据 url 中的 scope 参数决定服务导出方式，分别如下：

scope = none，不导出服务
scope != remote，导出到本地
scope != local，导出到远程

这里贴一下第三种情况的源码:

//如果配置不是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);}
}

Invoker 创建过程
Invoker 是实体域，它是 Dubbo 的核心模型，其它模型都向它靠扰，或转换成它，它代表一个可执行体，可向它发起 invoke 调用，它有可能是一个本地的实现，也可能是一个远程的实现，也可能一个集群实现
com.alibaba.dubbo.rpc.proxy.javassist.JavassistProxyFactory.getInvoker

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {// 为目标类创建 Wrapperfinal Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);// 创建匿名 Invoker 类对象，并实现 doInvoke 方法。return new AbstractProxyInvoker<T>(proxy, type, url) {@Overrideprotected Object doInvoke(T proxy, String methodName,Class<?>[] parameterTypes,Object[] arguments) throws Throwable {// 调用 Wrapper 的 invokeMethod 方法，invokeMethod 最终会调用目标方法return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);}};
}

代码生成完毕后，通过 Javassist 生成 Class 对象，最后再通过反射创建 Wrapper 实例。

com.alibaba.dubbo.common.bytecode.Wrapper.makeWrapper
（http://dubbo.apache.org/zh-cn/docs/source_code_guide/export-service.html 搜 2.2.1）
动态创建类的过程就不分析了。
通过 ClassGenerator 为刚刚生成的代码构建 Class 类，并通过反射创建对象。ClassGenerator 是 Dubbo 自己封装的，该类的核心是 toClass() 的重载方法 toClass(ClassLoader, ProtectionDomain)，该方法通过 javassist 构建 Class。

服务导出到本地：
com.alibaba.dubbo.config.ServiceConfig.exportLocal

@SuppressWarnings({"unchecked", "rawtypes"})
private void exportLocal(URL url) {if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {URL local = URL.valueOf(url.toFullString()).setProtocol(Constants.LOCAL_PROTOCOL).setHost(LOCALHOST).setPort(0);ServiceClassHolder.getInstance().pushServiceClass(getServiceClass(ref));Exporter<?> exporter = protocol.export(proxyFactory.getInvoker(ref, (Class) interfaceClass, local));exporters.add(exporter);logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry");}
}

导出服务到远程：

com.alibaba.dubbo.registry.integration.RegistryProtocol

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {//export invokerfinal ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);//registry providerfinal 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);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. 调用 doLocalExport 导出服务
2. 向注册中心注册服务
3. 向注册中心进行订阅 override 数据
4. 创建并返回 DestroyableExporter

主要分析下服务导出：

com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol.export

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {URL url = invoker.getUrl();// 获取服务标识，理解成服务坐标也行。由服务组名，服务名，服务版本号以及端口组成。比如：// demoGroup/com.alibaba.dubbo.demo.DemoService:1.0.1:20880String key = serviceKey(url);// 创建 DubboExporterDubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);// 将 <key, exporter> 键值对放入缓存中exporterMap.put(key, exporter);// 本地存根相关代码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) {// 省略日志打印代码} else {stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);}}// 启动服务器openServer(url);// 优化序列化optimizeSerialization(url);return exporter;
}

private ExchangeServer createServer(URL url) {url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY,// 添加心跳检测配置到 url 中url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));// 获取 server 参数，默认为 nettyString str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);// 通过 SPI 检测是否存在 server 参数所代表的 Transporter 拓展，不存在则抛出异常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, DubboCodec.NAME);ExchangeServer server;try {// 创建 ExchangeServerserver = Exchangers.bind(url, requestHandler);} catch (RemotingException e) {throw new RpcException("Fail to start server...");}// 获取 client 参数，可指定 netty，minastr = url.getParameter(Constants.CLIENT_KEY);if (str != null && str.length() > 0) {// 获取所有的 Transporter 实现类名称集合，比如 supportedTypes = [netty, mina]Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();// 检测当前 Dubbo 所支持的 Transporter 实现类名称列表中，// 是否包含 client 所表示的 Transporter，若不包含，则抛出异常if (!supportedTypes.contains(str)) {throw new RpcException("Unsupported client type...");}}return server;
}

createServer 包含三个核心的逻辑。
第一是检测是否存在 server 参数所代表的 Transporter 拓展，不存在则抛出异常。
第二是创建服务器实例。
第三是检测是否支持 client 参数所表示的 Transporter 拓展，不存在也是抛出异常。
两次检测操作所对应的代码比较直白了，无需多说。

server = Exchangers.bind(url, requestHandler);

看下 bind 方法,最终构建netty容器，开放链接。

com.alibaba.dubbo.remoting.exchange.Exchangers

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");// 获取 Exchanger，默认为 HeaderExchanger。// 紧接着调用 HeaderExchanger 的 bind 方法创建 ExchangeServer 实例return getExchanger(url).bind(url, handler);
}

com.alibaba.dubbo.remoting.Transporters

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 {// 如果 handlers 元素数量大于1，则创建 ChannelHandler 分发器handler = new ChannelHandlerDispatcher(handlers);}// 获取自适应 Transporter 实例，并调用实例方法return getTransporter().bind(url, handler);
}

com.alibaba.dubbo.remoting.transport.netty.NettyServer

public NettyServer(URL url, ChannelHandler handler) throws RemotingException {// 调用父类构造方法super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}

@Override
protected void doOpen() throws Throwable {NettyHelper.setNettyLoggerFactory();// 创建 boss 和 worker 线程池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));// 创建 ServerBootstrapbootstrap = new ServerBootstrap(channelFactory);final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);channels = nettyHandler.getChannels();bootstrap.setOption("child.tcpNoDelay", true);// 设置 PipelineFactorybootstrap.setPipelineFactory(new ChannelPipelineFactory() {@Overridepublic ChannelPipeline getPipeline() {NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);ChannelPipeline pipeline = Channels.pipeline();pipeline.addLast("decoder", adapter.getDecoder());pipeline.addLast("encoder", adapter.getEncoder());pipeline.addLast("handler", nettyHandler);return pipeline;}});// 绑定到指定的 ip 和端口上channel = bootstrap.bind(getBindAddress());
}

dubbo 默认使用的 NettyServer 是基于 netty 3.x 版本实现的，比较老了。
因此 Dubbo 另外提供了 netty 4.x 版本的 NettyServer，大家可在使用 Dubbo 的过程中按需进行配置。

再分析下服务注册：

服务注册操作对于 Dubbo 来说不是必需的，通过服务直连的方式就可以绕过注册中心。
但通常我们不会这么做，直连方式不利于服务治理，仅推荐在测试服务时使用。对于 Dubbo 来说，注册中心虽不是必需，但却是必要的。

获取注册中心
com.alibaba.dubbo.registry.support.AbstractRegistryFactory.getRegistry

@Override
public Registry getRegistry(URL url) {url = url.setPath(RegistryService.class.getName()).addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName()).removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);String key = url.toServiceString();LOCK.lock();try {// 访问缓存Registry registry = REGISTRIES.get(key);if (registry != null) {return registry;}// 缓存未命中，创建 Registry 实例registry = createRegistry(url);if (registry == null) {throw new IllegalStateException("Can not create registry...");}// 写入缓存REGISTRIES.put(key, registry);return registry;} finally {LOCK.unlock();}
}

先创建注册中心实例，之后再通过注册中心实例注册服务。

com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistry.doRegister

@Override
protected void doRegister(URL url) {try {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);}
}

zkClient.create()方法是根据 CuratorFrameWork 与zk连接的。

dubbo源码解析（二）相关推荐

dubbo源码解析（九）远程通信——Transport层
远程通讯--Transport层目标:介绍Transport层的相关设计和逻辑.介绍dubbo-remoting-api中的transport包内的源码解析. 前言先预警一下,该文篇幅会很长,做好 ...
dubbo源码解析（十）远程通信——Exchange层
远程通讯--Exchange层目标:介绍Exchange层的相关设计和逻辑.介绍dubbo-remoting-api中的exchange包内的源码解析. 前言上一篇文章我讲的是dubbo框架设计中 ...
dubbo(5) Dubbo源码解析之服务调用过程
来源:https://juejin.im/post/5ca4a1286fb9a05e731fc042 Dubbo源码解析之服务调用过程简介在前面的文章中,我们分析了 Dubbo SPI.服务导出与 ...
dubbo源码解析之框架粗谈
dubbo框架设计一.dubbo框架整体设计二.各层说明三.dubbo工程模块分包四.依赖关系五.调用链文章系列 [一.dubbo源码解析之框架粗谈] [二.dubbo源码解析之dubbo ...
Dubbo源码解析 --- DIRECTORY和ROUTER
Dubbo源码解析 --- DIRECTORY和ROUTER 今天看一下Directory和Router. 我们直接从代码看起(一贯风格),先看后总结,对着总结再来看,相信会收获很多.我们先看com. ...
dubbo源码解析-集群容错架构设计
前言本来是想把整个dubbo源码解析一次性弄完,再做成一个系列来发布的,但是正巧最近有位好朋友要去杭州面试,就和我交流了一下.本着对dubbo源码略有心得的心态,在交流过程中也发表了个人的一些粗劣的 ...
dubbo源码解析-zookeeper创建节点
前言在之前dubbo源码解析-本地暴露中的前言部分提到了两道高频的面试题,其中一道dubbo中zookeeper做注册中心,如果注册中心集群都挂掉,那发布者和订阅者还能通信吗?在上周的dubbo源码 ...
dubbo源码解析-逻辑层设计之服务降级
Dubbo源码解析系列文章均来自肥朝简书前言在dubbo服务暴露系列完结之后,按计划来说是应该要开启dubbo服务引用的讲解.但是现在到了年尾,一些朋友也和我谈起了明年跳槽的事.跳槽这件事,无非也 ...
Dubbo源码解析 —— Router
作者:肥朝原文地址:http://www.jianshu.com/p/278e782eef85 友情提示:欢迎关注公众号[芋道源码].????关注后,拉你进[源码圈]微信群和[肥朝]搞基嗨皮. 友情 ...

dubbo源码解析（二）

1.服务导出

dubbo源码解析（二）相关推荐

最新文章

热门文章