包装设计征集网站,包头市住房与城乡建设部网站,神经网络设计,浙江省外贸公司排名大家好#xff0c;我是烤鸭#xff1a; dubbo 源码解析#xff1a; 1.服务导出 介绍: Dubbo 服务导出过程始于 Spring 容器发布刷新事件#xff0c;Dubbo 在接收到事件后#xff0c;会立即执行服务导出逻辑。整个逻辑大致可分为三个部分#xff0c;第一部分是前置工作我是烤鸭 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 InvokerT getInvoker(T proxy, ClassT type, URL url) {// 为目标类创建 Wrapperfinal Wrapper wrapper Wrapper.getWrapper(proxy.getClass().getName().indexOf($) 0 ? proxy.getClass() : type);// 创建匿名 Invoker 类对象并实现 doInvoke 方法。return new AbstractProxyInvokerT(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 ExporterT export(final InvokerT originInvoker) throws RpcException {//export invokerfinal ExporterChangeableWrapperT 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 ExporterT() {public InvokerT 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 ExporterT export(InvokerT invoker) throws RpcException {URL url invoker.getUrl();// 获取服务标识理解成服务坐标也行。由服务组名服务名服务版本号以及端口组成。比如// demoGroup/com.alibaba.dubbo.demo.DemoService:1.0.1:20880String key serviceKey(url);// 创建 DubboExporterDubboExporterT exporter new DubboExporterT(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 参数可指定 nettyminastr url.getParameter(Constants.CLIENT_KEY);if (str ! null str.length() 0) {// 获取所有的 Transporter 实现类名称集合比如 supportedTypes [netty, mina]SetString 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连接的。