Dubbo底层源码解析

1、Dubbo核心模块职责介绍

2、源码基础

2.1 SPI自适应

2.2 Netty实现简易Dubbo

3、标签解析

3.1 标签属性配置

3.2 配置渊源

4、服务暴露

4.1 服务暴露启动逻辑

4.2 参数检查

4.3 组装url

4.4 服务暴露分类

4.5 服务暴露

4.6 服务注册

5、服务发现

5.1 服务发现原理

5.2 服务引用入口

5.3 创建代理

5.4 代理调用

1、Dubbo核心模块职责介绍

Dubbo发展到现在，已经有3.0版本了，但是官方建议应用于生产的最高版本还是2.7，所以我们以下课程都是以2.7作为标准！

源码下载地址：https://github.com/apache/dubbo/tree/2.7.x；该源码项目是完全基于MAVEN开发的，下载解压后直接导入idea即可，下载相关依赖包会比较久

dubbo common 通用模块，定义了几乎所有dubbo模块都会使用到的一些通用与业务领域无关的工具类（IO处理、日志处理、配置处理、类处理等等），线程池扩展、二进制代码处理、class编译处理、JSON处理、数据存储接口，系统版本号等等通用的类和接口。 dubbo rpc 分布式服务框架的核心是RPC，这是最基本的功能，这个模块定义了rpc的一些抽象的rpc接口和实现类，包括服务发布，服务调用代理，远程调用结果及异常，RPC调用网络协议，RPC调用监听器和过滤器等等。该模块提供了默认的基于dubbo协议的实现模块，还提供了hessian、http、rest、rmi、thrift和webservice等协议的实现，还实现了injvm的本地调用实现，灵活性强，非常通用，能够满足绝大多数项目的使用需求，而且还可以自行实现RPC协议。 dubbo registry  注册中心也是最重要的组成部分，它是RPC中的consumer和provider两个重要角色的协调者。该项目定义了核心的注册中心接口和实现。具体实现留给了其它项目。有一个默认的实现模块，注册中心提供了mutilcast、redis和zookeeper等多种方式的注册中心实现，用于不同的使用场景。 dubbo remoting 该模块是dubbo中的远程通讯模块。RPC的实现基础就是远程通讯，consmer要调用provider的远程方法必须通过网络远程通讯实现。该模块定义了远程传输器、终端（endpoint）、客户端、服务端、编码解码器、数据交换、缓冲区、通讯异常定义等等核心的接口及类构成。他是对于远程网络通讯的抽象。提供了诸如netty、mina、grizzly、http、p2p和zookeeper的协议和技术框架的实现方式。 dubbo monitor  该模块是dubbo的监控模块，通过该模块可以监控服务调用的各种信息，例如调用耗时、调用量、调用结果等等，监控中心在调用过程中收集调用的信息，发送到监控服务，在监控服务中可以存储这些信息，对这些数据进行统计分析，最终可以产生各种维护的调用监控信息。dubbo默认提供了一个实现，该实现非常简单，只是作为默认的实现范例，生产环境使用价值不高，需要自行实现自己的监控。 dubbo container  dubbo服务运行容器api模块。定义了启动容器列表的包含应用程序入口main方法的类Main；定义了容器接口Container，该接口包含了启动和停止方法定义；还有一些通用的分页功能的相关类。dubbo内置了javaconfig、jetty、log4j、logback和spring几种容器的实现。 dubbo config  该模块依赖了几乎所有的其它模块，他是dubbo的配置模块，通过它的配置和组装将dubbo组件的多个模块整合在一起给最终的开发者提供有价值的分布式服务框架。通过它的配置可以让开发者选择符合自己需求和使用场景的模块和技术，它定义了面向dubbo使用者的各种信息配置，比如服务发布配置、方法发布配置、服务消费配置、应用程序配置、注册中心配置、协议配置、监控配置等等。另外还有一个spring的配置模块，定义了一些spring的XML Schema，能够大大简化使用dubbo的配置，可以大大降低spring使用场景的学习和配置成本。 dubbo cluster 该模块是dubbo实现的集群模块。支持远程服务的集群，支持多种集群调用策略，包括failover,failsafe,failfast,failback,forking等。并且支持目录服务，注册中心就是目录服务的一种实现，支持负载均衡，该模块还实现了路由器特性，此外还包括合并技术，当将调用请求分发给所有的服务提供者，则会返回多个结果，则将多个结果合并需要用到合并器的实现，该模块也是非常有个性的一个模块。 dubbo admin 该项目是一个web应用，可以独立部署，它可以管理dubbo服务，通过该管理应用可以连接注册中心，重点是读取注册中心中的信息，也可以通过该应用改写注册中心的信息，从而实现动态的管控服务。该模块的功能也非常简单，对于实际的生产使用场景，还需要对该应用的功能进行扩展和定制，以满足实际的使用场景。

2、源码基础

我们需要先学习几个基本的技能点，后续解读源码时才能事半功倍游刃有余，

2.1 SPI自适应

2.1.1 SPI简介

SPI 全称为 Service Provider Interface，是一种服务发现机制。SPI 的本质是将接口实现类的全限定名配置在文件中，并由服务加载器读取配置文件，加载实现类。这样可以在运行时，动态为接口替换实现类。正因此特性，我们可以很容易的通过 SPI 机制为我们的程序提供拓展功能。SPI 机制在第三方框架中也有所应用，比如 Dubbo 就是通过 SPI 机制加载所有的组件。不过，Dubbo 并未使用 Java 原生的 SPI 机制，而是对其进行了增强，使其能够更好的满足需求。在 Dubbo 中，SPI 是一个非常重要的模块。基于 SPI，我们可以很容易的对 Dubbo 进行拓展。如果大家想要学习 Dubbo 的源码，SPI 机制务必弄懂。接下来，我们先来了解一下 Java SPI 与 Dubbo SPI 的用法，然后再来分析 Dubbo SPI 的源码。

2.1.2 Java SPI实例

package com.ydt.java.spi;

/*** 汽车接口类*/
public interface Car {void run();
}

package com.ydt.java.spi;

/*** 汽车接口实现类：奥迪*/
public class AudiCar implements Car{public void run() {System.out.println("我是奥迪.....，精英专用！！！");}
}

package com.ydt.java.spi;

/*** 汽车接口实现类：宝马*/
public class BMWCar implements Car{public void run() {System.out.println("我是宝马.....，暴发户专用！！！");}
}

接下来 resources/META-INF/services 文件夹下创建一个文件，名称为Car接口的全限定名 com.ydt.java.spi.Car。文件内容为Car实现类的全限定的类名，如下：

//测试代码
@Test
public void test1(){ServiceLoader<Car> serviceLoader = ServiceLoader.load(Car.class);Iterator<Car> iterator = serviceLoader.iterator();while (iterator.hasNext()){Car car = iterator.next();if("com.ydt.java.spi.BMWCar".equals(car.getClass().getName())){car.run();}}
}

2.1.3 Dubbo SPI实例

Dubbo 并未使用 Java SPI，而是重新实现了一套功能更强的 SPI 机制。Dubbo SPI 的相关逻辑被封装在了 ExtensionLoader（扩展点加载器）类中，通过 ExtensionLoader，我们可以加载指定的实现类。

与 Java SPI 实现类配置不同，Dubbo SPI 是通过键值对的方式进行配置，这样我们可以按需加载指定的实现类。另外，在测试 Dubbo SPI 时，需要在 Car接口上标注 @SPI 注解

    @Testpublic void test2(){ExtensionLoader<Car> extensionLoader =ExtensionLoader.getExtensionLoader(Car.class);Car car = extensionLoader.getExtension("bmw");//指定加载car.run();}

2.1.4 Java SPI vs Dubbo SPI

首先很明显的一点：Dubbo SPI支持按需加载接口实现类，节省了资源开支，并且不要像Java SPI那样使用实现类的限定名来判断的方式（不觉得很不优雅吗？）

其次还增加了 IOC 和 AOP 等特性，Spring拥有IOC和AOP的特性，而不能说IOC和AOP只有Spring才拥有，他们两个只是一个概念而已：

2.1.4.1 Dubbo SPI实现IOC

使用Dubbo SPI实现IOC，需要引入一个URL（服务总线）的概念，其中包括我们后面要用到的很多关于Dubbo应用配置，如协议，端口，参数配置等！

而我们的Dubbo源码里面一个重要的地方就是围绕URL服务总线进行解析和应用！

1、修改之前Dubbo SPI实例，在Car接口run()方法中增加URL参数，并且添加@Adaptive注解。它的实现类也相应的增加该参数

package com.ydt.java.spi;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Adaptive;
import org.apache.dubbo.common.extension.SPI;

/*** 汽车接口类* @SPI 可以设置默认值，如果使用默认值那么在按需加载的地方传参为"true"即可*/
@SPI
public interface Car {//增加汽车类型配置，用于决定使用哪个实现类加载，@Adaptive注解作用可以参照源码@Adaptive(value = "carType")void run(URL url);
}

2、增加一个司机类用于注入汽车类

package com.ydt.java.spi;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.SPI;

@SPI("old")
public interface Driver {public void driverCar(URL url);
}

package com.ydt.java.spi;


import org.apache.dubbo.common.URL;


public class OldDriver implements Driver {
private Car car;
public void setCar(Car car) {this.car = car;}
public void driverCar(URL url){System.out.println("哥哥考了驾照十年了......");car.run(url);}
}

配置SPI加载映射：

测试调用：

@Testpublic void test3(){ExtensionLoader<Driver> extensionLoader =ExtensionLoader.getExtensionLoader(Driver.class);Driver driver = extensionLoader.getDefaultExtension();Map<String,String> map = new HashMap<>();map.put("carType","bmw");URL url = new URL(null,null,0,map);driver.driverCar(url);}
}

2.1.4.1 Dubbo SPI实现AOP

在之前的Dubbo SPI实例的基础上，增加一个汽车包装类，在run方法前后执行一些业务逻辑而对于用户来说是透明无感知的！

package com.ydt.java.spi;

public class CarWrapper implements Car {private Car car;
public CarWrapper(Car car) {this.car = car;}
@Overridepublic void run() {System.out.println("哥哥存钱喽.........");car.run();System.out.println("哥哥买车喽.........");}
}

然后按Java SPI的配置方式配置包装类全限定名

继续测试：

对于SPI配置红色告警，那是因为现在Dubbo推荐使用的目录为resources/META-INF/dubbo/internal

2.1.5 Dubbo SPI源码解析

根据上面的案例，我们知道首先通过 ExtensionLoader的getExtensionLoader方法获取一个ExtensionLoader实例，然后再通过ExtensionLoader的getExtension 方法获取拓展类对象

public T getExtension(String name) {if (StringUtils.isEmpty(name)) {throw new IllegalArgumentException("Extension name == null");}if ("true".equals(name)) {return getDefaultExtension();}final Holder<Object> holder = getOrCreateHolder(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;
}

创建拓展对象过程

private T createExtension(String name, boolean wrap) {//从配置文件中加载所有的拓展类，可得到“配置项名称”到“配置类”的映射关系表Class<?> clazz = getExtensionClasses().get(name);if (clazz == null) {throw findException(name);}try {T instance = (T) EXTENSION_INSTANCES.get(clazz);if (instance == null) {// 通过反射创建实例EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.newInstance());instance = (T) EXTENSION_INSTANCES.get(clazz);}//向实例中注入依赖，完成IOCinjectExtension(instance);
if (wrap) {List<Class<?>> wrapperClassesList = new ArrayList<>();if (cachedWrapperClasses != null) {wrapperClassesList.addAll(cachedWrapperClasses);wrapperClassesList.sort(WrapperComparator.COMPARATOR);Collections.reverse(wrapperClassesList);}
if (CollectionUtils.isNotEmpty(wrapperClassesList)) {// 循环创建包装实例，完成AOPfor (Class<?> wrapperClass : wrapperClassesList) {Wrapper wrapper = wrapperClass.getAnnotation(Wrapper.class);if (wrapper == null|| (ArrayUtils.contains(wrapper.matches(), name) && !ArrayUtils.contains(wrapper.mismatches(), name))) {// 将当前 instance 作为参数传给 Wrapper 的构造方法，并通过反射创建 Wrapper 实例。// 然后向 Wrapper 实例中注入依赖，最后将 Wrapper 实例再次赋值给 instance 变量instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance));}}}}
initExtension(instance);return instance;} catch (Throwable t) {throw new IllegalStateException("Extension instance (name: " + name + ", class: " +type + ") couldn't be instantiated: " + t.getMessage(), t);}
}

createExtension 方法的逻辑稍复杂一下，包含了如下的步骤：

通过 getExtensionClasses 获取所有的拓展类
通过反射创建拓展对象
向拓展对象中注入依赖
将拓展对象包裹在相应的 Wrapper 对象中，通过包装类完成AOP

以上步骤中，第一个步骤是加载拓展类的关键，第三和第四个步骤是 Dubbo IOC 与 AOP 的具体实现

接下来我们来看看这四步Dubbo SPI怎么处理的：

1、获取所有的拓展类

private Map<String, Class<?>> getExtensionClasses() {// 从缓存中获取已加载的拓展类Map<String, Class<?>> classes = cachedClasses.get();//双重检查if (classes == null) {synchronized (cachedClasses) {classes = cachedClasses.get();if (classes == null) {//加载拓展类classes = loadExtensionClasses();//将拓展类加入缓存cachedClasses.set(classes);}}}return classes;
}

这里也是先检查缓存，若缓存未命中，则通过 synchronized 加锁。加锁后再次检查缓存，并判空。此时如果 classes 仍为 null，则通过 loadExtensionClasses 加载拓展类。下面分析 loadExtensionClasses 方法的逻辑。

加载拓展类

private Map<String, Class<?>> loadExtensionClasses() {//处理@SPI注解内容解析cacheDefaultExtensionName();Map<String, Class<?>> extensionClasses = new HashMap<>();for (LoadingStrategy strategy : strategies) {// 加载指定文件夹下的配置文件，strategy.directory()分别包括dubbo,dubbo/internal,servicesloadDirectory(extensionClasses, strategy.directory(), type.getName(), strategy.preferExtensionClassLoader(), strategy.overridden(), strategy.excludedPackages());loadDirectory(extensionClasses, strategy.directory(), type.getName().replace("org.apache", "com.alibaba"), strategy.preferExtensionClassLoader(), strategy.overridden(), strategy.excludedPackages());}return extensionClasses;
}

解析SPI注解，获取扩展类的配置，如：@SPI("bmw")

private void cacheDefaultExtensionName() {// 获取 SPI 注解，这里的 type 变量是在调用 getExtensionLoader 方法时传入的final SPI defaultAnnotation = type.getAnnotation(SPI.class);if (defaultAnnotation == null) {return;}
String value = defaultAnnotation.value();if ((value = value.trim()).length() > 0) {// 对 SPI 注解内容进行逗号切分String[] names = NAME_SEPARATOR.split(value);// 检测 SPI 注解内容是否合法，不合法则抛出异常，其实就是说只能配一个.....if (names.length > 1) {throw new IllegalStateException("More than 1 default extension name on extension " + type.getName()+ ": " + Arrays.toString(names));}// 设置默认名称，参考 getDefaultExtension 方法if (names.length == 1) {cachedDefaultName = names[0];}}
}

回过头来看下IOC注入的方法：

private T injectExtension(T instance) {
if (objectFactory == null) {return instance;}
try {// 遍历目标类的所有方法for (Method method : instance.getClass().getMethods()) {//检测方法是否以 set 开头，且方法仅有一个参数，且方法访问级别为 publicif (!isSetter(method)) {continue;}/*** Check {@link DisableInject} to see if we need auto injection for this property*/if (method.getAnnotation(DisableInject.class) != null) {continue;}// 获取 setter 方法参数类型Class<?> pt = method.getParameterTypes()[0];if (ReflectUtils.isPrimitives(pt)) {continue;}
try {// 获取属性名，比如 setName 方法对应属性名 nameString property = getSetterProperty(method);// 从 ObjectFactory 中获取依赖对象Object object = objectFactory.getExtension(pt, property);if (object != null) {// 通过反射调用 setter 方法设置依赖method.invoke(instance, object);}} catch (Exception e) {logger.error("Failed to inject via method " + method.getName()+ " of interface " + type.getName() + ": " + e.getMessage(), e);}
}} catch (Exception e) {logger.error(e.getMessage(), e);}return instance;
}

有些同学问到，我们设置的URL参数呢：

其实这是Dubbo SPI一种自适应扩展机制，跟进上面IOC注入的代码：

调用链路太长，大家可以根据debug一直跟进到如下方法

public String generate() {// no need to generate adaptive class since there's no adaptive method found.if (!hasAdaptiveMethod()) {throw new IllegalStateException("No adaptive method exist on extension " + type.getName() + ", refuse to create the adaptive class!");}
StringBuilder code = new StringBuilder();//包名code.append(generatePackageInfo());//导入类code.append(generateImports());//类定义code.append(generateClassDeclaration());Method[] methods = type.getMethods();for (Method method : methods) {//方法定义code.append(generateMethod(method));}//结尾code.append("}");
if (logger.isDebugEnabled()) {logger.debug(code.toString());}return code.toString();
}

把这段代码生成的字符串拷贝出去，他完全就是一个适配器对象类！如下：

package com.ydt.java.spi;

import org.apache.dubbo.common.extension.ExtensionLoader;

public class Car$Adaptive implements com.ydt.java.spi.Car {public void run(org.apache.dubbo.common.URL arg0) {if (arg0 == null) throw new IllegalArgumentException("url == null");org.apache.dubbo.common.URL url = arg0;String extName = url.getParameter("carType");if (extName == null)throw new IllegalStateException("Failed to get extension (com.ydt.java.spi.Car) name from url (" + url.toString() + ") use keys([carType])");//这个地方是不是很熟悉啊，根据url参数重新去加载对应的实现类对象！com.ydt.java.spi.Car extension = (com.ydt.java.spi.Car) ExtensionLoader.getExtensionLoader(com.ydt.java.spi.Car.class).getExtension(extName);extension.run(arg0);}
}

Dubbo 会为拓展接口生成具有代理功能的代码（适配器对象类），然后通过 javassist 或 jdk 编译这段代码，得到 Class 类。最后再通过反射创建代理类对象，最后再回过头来注入到原对象；整个过程比较复杂，我们理解其原理就行

2.2 Netty实现简易Dubbo

2.2.1 Netty服务模型图

2.2.2 Netty简单demo

发布的业务服务

package com.ydt.netty.inter;

public interface HelloService {
String hello(String ping);

}

package com.ydt.netty.provider;

import com.ydt.netty.inter.HelloService;

/*** 实现类*/
public class HelloServiceImpl implements HelloService {
@Overridepublic String hello(String msg) {System.out.println("收到客户端消息: " + msg);return "收到服务端信息： " + msg;}
}

ChannelHandler服务端业务处理

package com.ydt.netty.handler;

import com.ydt.netty.service.HelloServiceImpl;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/*** 用于处理请求数据*/
public class ServerStub extends ChannelInboundHandlerAdapter {
@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {String message = msg.toString();//人造协议处理if (message.split("&")[0].equals("dubbo")){String result = new HelloServiceImpl().hello(message.split("&")[1]);ctx.writeAndFlush(result);}else {ctx.writeAndFlush("不支持该协议！");}
}
}

Netty服务端：

package com.ydt.netty.provider;

import com.ydt.netty.handler.ServerStub;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/*** 服务端*/
public class NettyServer {
public static void main(String[] args) {startServer("127.0.0.1", 7002);}
/*** 启动客户端*/public static void startServer(String hostName, int port) {try {ServerBootstrap bootstrap = new ServerBootstrap();NioEventLoopGroup mainLoopGroup = new NioEventLoopGroup(1);NioEventLoopGroup workLoopGroup = new NioEventLoopGroup();bootstrap.group(mainLoopGroup,workLoopGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline p = ch.pipeline();p.addLast(new StringDecoder());//解码器p.addLast(new StringEncoder());//编码器p.addLast(new ServerStub());//业务处理器提供方Stub}});bootstrap.bind(hostName, port).await();} catch (InterruptedException e) {e.printStackTrace();}}
}

ChannelHandler客户端业务处理

package com.ydt.netty.handler;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.concurrent.Callable;

public class ClientStub extends ChannelInboundHandlerAdapter implements Callable {
private ChannelHandlerContext context;
private String result;//客户端调用后,返回的结果
private String para;//客户端调用方法时传递进来的参数
/*** 与服务器的连接已经建立之后将被调用，ctx传递给context,做成属性** @param ctx*/@Overridepublic void channelActive(ChannelHandlerContext ctx) {//其它方法里面 使用到当前Handler里面的上下文,因此需要保存下上下文当成属性context = ctx;}
/*** 收到服务端数据，唤醒等待线程，与Callable实现方法call同步，需要加synchronized*/@Overridepublic synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {result = msg.toString();notify();}
/*** 数据发出，等待channelRead处理后被唤醒，与*  ChannelInboundHandlerAdapter实现方法channelRead同步，需要加synchronized*/@Overridepublic synchronized Object call() throws InterruptedException {context.writeAndFlush(para);//如果不加该阻塞，服务器返回数据为空或者多个返回会一起设置到resultwait();return result;}
public void setPara(String para) {this.para = para;}
}

Netty客户端：

package com.ydt.netty.consumer;

import com.ydt.netty.handler.ClientStub;
import com.ydt.netty.service.HelloService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NettyClient {
private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
private static ClientStub client;//自定义协议头private static String protocal = "dubbo";
public static void main(String[] args) throws InterruptedException {// 创建一个代理对象HelloService service = (HelloService) getBean(HelloService.class,protocal);while (true) {System.out.println(service.hello("你好 dubbo ~ "));Thread.sleep(2000);}}
/*** 创建一个代理对象*/public static Object getBean(final Class<?> serviceClass,String protocal) {return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),//类加载器，写活，不要写成目标接口类加载器new Class<?>[]{serviceClass},new InvocationHandler() {//proxy：代理对象本身，method: 代理对象方法，args: 将来传递的参数-->‘协议头’ 传递进来的信息public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {if (client == null) {initClient();}// 设置参数值，包括协议，具体可以参考zookeeper节点值方式client.setPara(protocal + "&" +args[0].toString());// submit(client)是一个Callable<T>对象---ClientStub// get()拿到ClientStub中实现的call方法的return的结果return executor.submit(client).get();}});}
/*** 初始化客户端*/private static void initClient() {client = new ClientStub();/*** 内部线程池，内部维护了一组线程，每个线程负责处理多个Channel上的事件，*  而一个Channel只对应于一个线程，这样可以回避多线程下的数据同步问题。*/EventLoopGroup group = new NioEventLoopGroup();//服务消费方BootstrapBootstrap b = new Bootstrap();b.group(group)//设置线程组.channel(NioSocketChannel.class)//使用NIO通道.handler(new ChannelInitializer<SocketChannel>() {//处理通道请求事件public void initChannel(SocketChannel ch) throws Exception {ChannelPipeline p = ch.pipeline();p.addLast(new StringDecoder());//解码器p.addLast(new StringEncoder());//编码器p.addLast(client);//业务处理器消费方Stub}});try {//建立长连接b.connect("127.0.0.1", 7002).await();} catch (InterruptedException e) {e.printStackTrace();}}
}

RPC调用流程说明

(1)服务消费方(Client)以本地调用方式调用服务

(2)client stub接收到调用后负责将方法、参数等封装成能够进行网络传输的消息体

(3)client stub将消息进行编码并发送到服务端

(4)server stub收到消息后进行解码

(5)server stub根据解码结果调用本地服务

(6)server stub将本地服务结果进行编码返回

(7)服务提供方（Server）将返回导入结果发送至服务消费方（Client）

(8)服务消费方（Client）接收到消息并进行解码

3、标签解析

3.1 标签属性配置

Dubbo 支持多种配置方式：

XML 配置：基于 Spring 的 Schema 和 XML 扩展机制实现
属性配置：加载 classpath 根目录下的 dubbo.properties
API 配置：通过硬编码方式配置（不推荐使用）
注解配置：通过注解方式配置（Dubbo-2.5.7及以上版本支持，不推荐使用）

对于 属性配置 方式，可以通过环境变量、-D 启动参数来指定 dubbo.properties 文件，加载文件顺序为：

-D 启动参数
环境变量
classpath 根目录（生产上默认使用这个就行了！）

/*-------------------------------------------ConfigUtils.java---------------------------------------------------*/
public static Properties getProperties() {if (PROPERTIES == null) {synchronized (ConfigUtils.class) {if (PROPERTIES == null) {//-D 启动参数String path = System.getProperty(CommonConstants.DUBBO_PROPERTIES_KEY);if (path == null || path.length() == 0) {//环境变量path = System.getenv(CommonConstants.DUBBO_PROPERTIES_KEY);if (path == null || path.length() == 0) {//classpath 根目录path = CommonConstants.DEFAULT_DUBBO_PROPERTIES;}}PROPERTIES = ConfigUtils.loadProperties(path, false, true);}}}return PROPERTIES;
}

可以通过服务调用链看到：

3.2 配置渊源

我们不会所有模式都会讲到，这里我们使用最常用的XML配置方式来作为讲解目标

3.2.1 命名空间和Schema约束

这一块跟Spring何其类似，说错了，应该说跟太多的开源框架都是大同小异！

我们可以随便找一个标签，一路链接过去：

需要注意的是dubbo很多标签属性并不是标签独有的，而是采用层次迭代方式，所以我们可以找到它的父类，看到一些共同属性：

3.2.2 解析xml

Dubbo 服务框架的 Schema 的解析通过 DubboNamespaceHandler 和 DubboBeanDefinitionParser 实现。

其中，DubboNamespaceHandler 扩展了 Spring 的 NamespaceHandlerSupport，通过重写它的 init() 方法给各个标签注册对应的解析器：

DubboBeanDefinitionParser 实现了 Spring 的 BeanDefinitionParser，通过重写 parse() 方法实现将标签解析为对应的 JavaBean

3.2.3 执行入口

那么问题来了，这个dubbo命名空间处理类DubboNamespaceHandler是怎么被调用的呢？

我们以前学过spring，知道spring有这么两个配置：

那么dubbo作为Spring的集大成者，他是不是也是这个思路呢，恭喜你，猜对了！

如果schema搞不太清楚又感兴趣的同学可以先移步到w3school学习下：Schema 教程

4、服务暴露

4.1 服务暴露启动逻辑

通过上一章节，我们知道dubbo:service 标签会被解析成 ServiceConfig(BeanDefine)，并最终会解析为其子类：ServiceBean

首先引入一个类：DubboBootstrap，主要处理dubbo所有的配置信息，功能主要有：

持有ConfigManager、Environment对象并且对其初始化，这两个对象都是与配置相关的；
更新配置中心配置对象ConfigCenterConfig的属性值；
加载元数据中心对象；
检查各个配置对象的属性值是否合法；
注册java的关闭钩子；
服务端服务的暴露。

其中最重要的就是服务端服务的暴露！

ServiceBean 实现了 InitializingBean，在类加载完成之后会调用 afterPropertiesSet() ，目前destroy方法啥也没干

@Override
public void afterPropertiesSet() throws Exception {if (StringUtils.isEmpty(getPath())) {if (StringUtils.isNotEmpty(getInterface())) {setPath(getInterface());}}
}

@Override
public void destroy() throws Exception {// no need to call unexport() here, see// org.apache.dubbo.config.spring.extension.SpringExtensionFactory.ShutdownHookListener
}

Dubbo从2.6.5开始，只使用延迟暴露，为什么？参考《Dubbo高级应用-分布式服务化》章节：延迟暴露，解决Spring初始化死锁问题

ServiceBean 扩展了 ServiceConfig，调用 export() 方法时由 ServiceConfig 完成服务暴露的功能实现。

/***********************************************ServiceConfig.java****************************************************/
public synchronized void export() {if (!shouldExport()) {return;}
if (bootstrap == null) {bootstrap = DubboBootstrap.getInstance();bootstrap.initialize();}//参数校验checkAndUpdateSubConfigs();
//init serviceMetadataserviceMetadata.setVersion(getVersion());serviceMetadata.setGroup(getGroup());serviceMetadata.setDefaultGroup(getGroup());serviceMetadata.setServiceType(getInterfaceClass());serviceMetadata.setServiceInterfaceName(getInterface());serviceMetadata.setTarget(getRef());//手动指定延迟时间，默认是spring容器初始化后正常速度暴露if (shouldDelay()) {DELAY_EXPORT_EXECUTOR.schedule(this::doExport, getDelay(), TimeUnit.MILLISECONDS);} else {//定时器中启动一个守护线程在 sleep 指定时间后再 doExportdoExport();}
exported();}

4.2 参数检查

在 ServiceConfig 的 checkAndUpdateSubConfigs() 方法中会进行参数检查和设置，包括：

泛化调用
本地实现
本地存根
本地伪装
配置（application、registry、protocol等）

/***********************************************ServiceConfig.java****************************************************/
private void checkAndUpdateSubConfigs() {// Use default configs defined explicitly with global scopecompleteCompoundConfigs();checkDefault();checkProtocol();// init some null configuration.List<ConfigInitializer> configInitializers = ExtensionLoader.getExtensionLoader(ConfigInitializer.class).getActivateExtension(URL.valueOf("configInitializer://"), (String[]) null);configInitializers.forEach(e -> e.initServiceConfig(this));
// if protocol is not injvm checkRegistryif (!isOnlyInJvm()) {checkRegistry();}this.refresh();
if (StringUtils.isEmpty(interfaceName)) {throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!");}//泛化调用if (ref instanceof GenericService) {interfaceClass = GenericService.class;if (StringUtils.isEmpty(generic)) {generic = Boolean.TRUE.toString();}} else {try {interfaceClass = Class.forName(interfaceName, true, Thread.currentThread().getContextClassLoader());} catch (ClassNotFoundException e) {throw new IllegalStateException(e.getMessage(), e);}checkInterfaceAndMethods(interfaceClass, getMethods());checkRef();generic = Boolean.FALSE.toString();}//本地实现if (local != null) {if ("true".equals(local)) {local = interfaceName + "Local";}Class<?> localClass;try {localClass = ClassUtils.forNameWithThreadContextClassLoader(local);} catch (ClassNotFoundException e) {throw new IllegalStateException(e.getMessage(), e);}if (!interfaceClass.isAssignableFrom(localClass)) {throw new IllegalStateException("The local implementation class " + localClass.getName() + " not implement interface " + interfaceName);}}//本地存根if (stub != null) {if ("true".equals(stub)) {stub = interfaceName + "Stub";}Class<?> stubClass;try {stubClass = ClassUtils.forNameWithThreadContextClassLoader(stub);} catch (ClassNotFoundException e) {throw new IllegalStateException(e.getMessage(), e);}if (!interfaceClass.isAssignableFrom(stubClass)) {throw new IllegalStateException("The stub implementation class " + stubClass.getName() + " not implement interface " + interfaceName);}}checkStubAndLocal(interfaceClass);//本地伪装ConfigValidationUtils.checkMock(interfaceClass, this);ConfigValidationUtils.validateServiceConfig(this);postProcessConfig();
}

4.3 组装url

在检查完参数之后，开始暴露服务。Dubbo 支持多协议和多注册中心：

/***********************************************ServiceConfig.java****************************************************/
private void doExportUrls() {//本地注册：缓存服务接口在本地，处理请求的时候直接根据消费者对方的URL直接匹配即可ServiceRepository repository = ApplicationModel.getServiceRepository();ServiceDescriptor serviceDescriptor = repository.registerService(getInterfaceClass());repository.registerProvider(getUniqueServiceName(),ref,serviceDescriptor,this,serviceMetadata);List<URL> registryURLs = ConfigValidationUtils.loadRegistries(this, true);//多协议、多注册中心for (ProtocolConfig protocolConfig : protocols) {String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version);// In case user specified path, register service one more time to map it to path.repository.registerService(pathKey, interfaceClass);// TODO, uncomment this line once service key is unifiedserviceMetadata.setServiceKey(pathKey);//针对每个协议、每个注册中心，开始组装 URL。doExportUrlsFor1Protocol(protocolConfig, registryURLs);}
}

上面的代码最后doExportUrlsFor1Protocol()方法完成功能：首先是将一些信息，比如版本、时间戳、方法名以及各种配置对象的字段信息放入到 map 中，map 中的内容将作为 URL 的查询字符串。构建好 map 后，紧接着是获取上下文路径、主机名以及端口号等信息。最后将 map 和主机名等数据传给 URL 构造方法创建 URL 对象。需要注意的是，这里出现的 URL 并非 java.net.URL，而是 com.alibaba.dubbo.common.URL

4.4 服务暴露分类

如果配置 scope=none，则不会进行服务暴露；

如果没有配置scope=remote，则会进行本地暴露。

如果没有配置scope=local，则会进行远程暴露。

    /***********************************************ServiceConfig.java******************************/       private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {//......................省略..........................// 如果没有配置local，进行远程暴露if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {if (CollectionUtils.isNotEmpty(registryURLs)) {//对所有的注册地址进行注册for (URL registryURL : registryURLs) {//如果协议是injvm,不进行注册，只供本地调用if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {continue;}//服务是否动态注册，如果设为false，注册后将显示后disable状态，需人工启用，并且服务提供者停止时，也不会自动取消注册，需人工禁                       用，默认true。url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY));//如果配置了dubbo:monitor,则加载到monitor监控URL monitorUrl = ConfigValidationUtils.loadMonitor(this, registryURL);if (monitorUrl != null) {url = url.addParameterAndEncoded(MONITOR_KEY, monitorUrl.toFullString());}if (logger.isInfoEnabled()) {if (url.getParameter(REGISTER_KEY, true)) {logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " +        registryURL);} else {logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);}}
// 查看是否设置了代理模式jdk or javassistString proxy = url.getParameter(PROXY_KEY);if (StringUtils.isNotEmpty(proxy)) {registryURL = registryURL.addParameter(PROXY_KEY, proxy);}//Dubbo SPI 生成动态代理执行器（ProxyFactory）,使用的javassist默认方式,比JDK方式性能要快Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass,                      registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString()));DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);//包装以后进行暴露到注册中心Exporter<?> exporter = PROTOCOL.export(wrapperInvoker);exporters.add(exporter);}  //......................省略..........................

4.5 服务暴露

在服务暴露时，有两种方式：

不使用注册中心：直接暴露对应协议的服务，引用服务时只能通过直连方式引用-----如DubboProtocol
使用注册中心：暴露对应协议的服务后，会将服务节点注册到注册中心，引用服务时可以通过注册中心动态获取服务提供者列表，也可以通过直连方式引用

--------如RegistryProtocol

/*********************************************RegistryProtocol.java*******************************************************/
/*** 1、调用 doLocalExport 导出服务* 2、向注册中心注册服务* 3、向注册中心进行订阅 override 数据* 4、创建并返回 DestroyableExporter* @param originInvoker* @param <T>* @return* @throws RpcException*/public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {//注册中心地址,如：zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-providerURL registryUrl = getRegistryUrl(originInvoker);// url to export locally//服务接口地址URL providerUrl = getProviderUrl(originInvoker);
// 订阅override数据，获取订阅 URL，如：provider://172.17.48.52:20880/com.alibaba.dubbo.demo.DemoService?category=configurators&check=false// 相同的注册中心配置多个相同的服务注册地址final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);//添加覆盖监听器，后加载的覆盖旧的服务地址overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);//暴露服务，调用的是具体的协议Protocol，如DubboProtocol，同样要执行无注册中心暴露逻辑final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);
//根据 URL 加载 Registry 实现类，比如 ZookeeperRegistryfinal Registry registry = getRegistry(originInvoker);//获取已注册的服务提供者 URL，如：dubbo://172.17.48.52:20880/com.alibaba.dubbo.demo.DemoService?anyhost=truefinal URL registeredProviderUrl = getUrlToRegistry(providerUrl, registryUrl);
// 获取 register 参数，根据 register 的值决定是否注册服务boolean register = providerUrl.getParameter(REGISTER_KEY, true);if (register) {// 向注册中心注册服务register(registryUrl, registeredProviderUrl);}
// 将注册的url加入到服务模型中// -----》private ConcurrentMap<String, ProviderModel> providers = new ConcurrentHashMap<>();registerStatedUrl(registryUrl, registeredProviderUrl, register);
//设置注册中心地址，如：zookeeper://192.168.223.128:2181/org.apache.dubbo.registry.RegistryService?application=demo-provider//设置订阅地址：provider://192.168.223.1:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=192.168.223.1&bind.port=20880exporter.setRegisterUrl(registeredProviderUrl);exporter.setSubscribeUrl(overrideSubscribeUrl);
// 向注册中心进行订阅 override 数据registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);//注册监听：RegistryProtocolListenernotifyExport(exporter);//Ensure that a new exporter instance is returned every time exportreturn new DestroyableExporter<>(exporter);}

不管是否使用了注册中心，服务暴露的逻辑都是一样，默认dubbo协议是使用的netty服务进行暴露，底层传输默认使用 NettyTransporter，最终是创建 NettyServer

/*********************************************RegistryProtocol.java*******************************************************/
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {// 访问缓存String key = getCacheKey(originInvoker);// computeIfAbsent：如果没有的话创建后写入bounds缓存return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(key, s -> {// 创建 Invoker 为委托类对象Invoker<?> invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);// 调用 protocol 的 export 方法导出服务，如果协议为dubbo，会进入DubboProtocol.export方法return new ExporterChangeableWrapper<>((Exporter<T>) protocol.export(invokerDelegate), originInvoker);});}

/*********************************************DubboProtocol.java*******************************************************/
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {//.....................................................//创建netty服务进行暴露openServer(url);optimizeSerialization(url);return exporter;
}

/*********************************************DubboProtocol.java*******************************************************/

private void openServer(URL url) {//获取 host:port，并将其作为服务器实例的 key，用于标识当前的服务器实例String key = url.getAddress();//client 也可以暴露一个只有server可以调用的服务。boolean isServer = url.getParameter(IS_SERVER_KEY, true);if (isServer) {// 访问缓存ProtocolServer server = serverMap.get(key);if (server == null) {synchronized (this) {server = serverMap.get(key);if (server == null) {// 创建服务器实例,如nettyserverMap.put(key, createServer(url));}}} else {//server支持reset,配合override功能使用server.reset(url);}}}

/*********************************************DubboProtocol.java*******************************************************/
private ProtocolServer createServer(URL url) {url = URLBuilder.from(url)//发送通道只读设置，避免篡改数据.addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())// 添加心跳检测配置到 url 中.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))// 添加编码解码器参数.addParameter(CODEC_KEY, DubboCodec.NAME).build();// 获取 server 参数，默认为 nettyString str = url.getParameter(SERVER_KEY, 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);}
ExchangeServer server;try {//创建 ExchangeServer，是不是跟它很像，没错就是它：bootstrap.bind(hostName, port).await();server = Exchangers.bind(url, requestHandler);} catch (RemotingException e) {throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);}// 获取 client 参数，默认nettystr = url.getParameter(CLIENT_KEY);if (str != null && str.length() > 0) {// 获取所有的 Transporter 实现类名称集合，比如 supportedTypes = [netty, mina]//是不是很眼熟啊，没错，就是SPISet<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();// 检测当前 Dubbo 所支持的 Transporter 实现类名称列表中，// 是否包含 client 所表示的 Transporter，若不包含，则抛出异常if (!supportedTypes.contains(str)) {throw new RpcException("Unsupported client type: " + str);}}//返回Dubbo支持协议的RPC服务端对象，默认NettyServerreturn new DubboProtocolServer(server);}
/*********************************************Exchangers.java*******************************************************/
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);}

/******************************一路跟踪bind，可以看到底层传输默认使用 NettyTransporter，最终是创建NettyServer*****************************/
public Server bind(URL url, ChannelHandler listener) throws RemotingException {return new NettyServer(url, listener);//netty进行服务的发布，请自行参考分布式netty课件
}

/******************************接下来跟踪NettyServer的父类，进入doOpen模板方法，开启NettyServer服务*****************************/protected void doOpen() throws Throwable {NettyHelper.setNettyLoggerFactory();// 创建 boss 和 worker 线程池//主线程组, 用于接受客户端的连接，但是不做任何处理，跟老板一样，不做事,这跟redis的主线程一样，只管连接ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));// 从线程组, 老板线程组会把任务丢给他，让手下线程组去做任务，这跟redis的同步分支线程一样，只管数据同步ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(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);bootstrap.setOption("backlog", getUrl().getPositiveParameter(BACKLOG_KEY, Constants.DEFAULT_BACKLOG));// 设置 PipelineFactory,是不是发现了什么bootstrap.setPipelineFactory(new ChannelPipelineFactory() {@Overridepublic 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;}});// // 绑定到指定的 ip 和端口上(很纯粹的Netty)channel = bootstrap.bind(getBindAddress());}

4.6 服务注册

如果使用了注册中心，则在通过具体协议（如 Dubbo 协议）暴露服务之后（即在 2.8 基础之上）进入服务注册流程，将服务节点注册到注册中心。

我们把目光再次移到 RegistryProtocol 的 export 方法上

/*********************************************RegistryProtocol.java*******************************************************/
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {//.......................................省略...............................................
// 获取 register 参数，根据 register 的值决定是否注册服务boolean register = providerUrl.getParameter(REGISTER_KEY, true);if (register) {// 向注册中心注册服务，看下面章节：服务注册register(registryUrl, registeredProviderUrl);}//.......................................省略...............................................}

/*********************************************RegistryProtocol.java*******************************************************/
private void register(URL registryUrl, URL registeredProviderUrl) {// 获取 RegistryRegistry registry = registryFactory.getRegistry(registryUrl);// 注册服务registry.register(registeredProviderUrl);}

本节内容以 Zookeeper 注册中心为例进行分析,下面先来看一下 getRegistry 方法的源码，这个方法由 AbstractRegistryFactory 实现

/*********************************************AbstractRegistryFactory.java*******************************************************/
public Registry getRegistry(URL url) {//....................................省略................................................String key = createRegistryCacheKey(url);// 锁定注册表访问进程以确保注册表的单个实例，也就是说不允许有多个注册中心实例对象LOCK.lock();try {// 访问缓存Registry registry = REGISTRIES.get(key);if (registry != null) {return registry;}//使用SPI扩展注册中心对象，点进去，这是一个模板方法，进入具体的子类，如ZookeeperRegistryFactoryregistry = createRegistry(url);if (registry == null) {throw new IllegalStateException("Can not create registry " + url);}// 写入缓存REGISTRIES.put(key, registry);return registry;} finally {// Release the lockLOCK.unlock();}}
/*********************************************ZookeeperRegistryFactory.java*******************************************************/public Registry createRegistry(URL url) {// 创建 ZookeeperRegistry，继续进去其构造方法return new ZookeeperRegistry(url, zookeeperTransporter);}

/*********************************************ZookeeperRegistry.java*******************************************************/
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {super(url);if (url.isAnyHost()) {throw new IllegalStateException("registry address == null");}// 获取组名，默认为 dubbo，就是咱们zookeeper管理控制台上那个根目录名称String group = url.getParameter(GROUP_KEY, DEFAULT_ROOT);if (!group.startsWith(PATH_SEPARATOR)) {group = PATH_SEPARATOR + group;}this.root = group;// 创建 Zookeeper 客户端，默认为 CuratorZookeeperTransporter，可以跟进去看看zkClient = zookeeperTransporter.connect(url);// 添加状态监听器，znode节点有变动得通知我zkClient.addStateListener((state) -> {if (state == StateListener.RECONNECTED) {logger.warn("Trying to fetch the latest urls, in case there're provider changes during connection loss.\n" +" Since ephemeral ZNode will not get deleted for a connection lose, " +"there's no need to re-register url of this instance.");//获取最新的服务地址ZookeeperRegistry.this.fetchLatestAddresses();} else if (state == StateListener.NEW_SESSION_CREATED) {logger.warn("Trying to re-register urls and re-subscribe listeners of this instance to registry...");try {//如果是监听session新建或者过期重连，进行服务地址覆盖ZookeeperRegistry.this.recover();} catch (Exception e) {logger.error(e.getMessage(), e);}} else if (state == StateListener.SESSION_LOST) {logger.warn("Url of this instance will be deleted from registry soon. " +"Dubbo client will try to re-register once a new session is created.");} else if (state == StateListener.SUSPENDED) {
} else if (state == StateListener.CONNECTED) {
}});}
/*********************************************ZookeeperRegistry.java*******************************************************/
public ZookeeperClient connect(URL url) {//.........................................省略..............................................// 避免创建过多的连接，所以添加锁,加上上面LOCK.lock()，双重锁，dubbo在这一块双重的很厉害synchronized (zookeeperClientMap) {if ((zookeeperClient = fetchAndUpdateZookeeperClientCache(addressList)) != null && zookeeperClient.isConnected()) {logger.info("find valid zookeeper client from the cache for address: " + url);return zookeeperClient;}
zookeeperClient = createZookeeperClient(url);logger.info("No valid zookeeper client found from cache, therefore create a new client for url. " + url);writeToClientMap(addressList, zookeeperClient);}return zookeeperClient;}
/*********************************************CuratorZookeeperTransporter.java*******************************************************/
public ZookeeperClient createZookeeperClient(URL url) {//zookeeperTransporter 类型为自适应拓展类，因此 connect 方法会在被调用时决定加载什么类型的 ZookeeperTransporter 拓展，默认为 CuratorZookeeperTransporterreturn new CuratorZookeeperClient(url);}

现在继续回过头跟进RegistryProtocol.register()方法，看后半场节点创建，一直跟到FailbackRegistry

/*********************************************FailbackRegistry.java*******************************************************/
public void register(URL url) {//dubbo:registry是否有属性集合配置分隔符，默认为“，”if (!acceptable(url)) {logger.info("URL " + url + " will not be registered to Registry. Registry " + url + " does not accept service of this protocol type.");return;}//将服务添加到注册集合中super.register(url);//移除重试注册失败（多次）和未重试注册失败（一次）的集合，避免重复注册removeFailedRegistered(url);removeFailedUnregistered(url);try {// 通过zkClient进行znode的创建doRegister(url);} catch (Exception e) {//...........................................省略............................................// 将注册失败的url添加到重试任务中，通过定时任务重新注册addFailedRegistered(url);}
}

/*********************************************ZookeeperRegistry.java*******************************************************/
protected void doRegister(URL url) {try {// 通过 Zookeeper 客户端创建节点，节点路径由 toUrlPath 方法生成，路径格式如下://   /${group}/${serviceInterface}/providers/${url}// 比如//   /dubbo/org.apache.dubbo.DemoService/providers/dubbo%3A%2F%2F127.0.0.1......zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));} catch (Throwable e) {throw new RpcException("Failed to register...");}
}

/*********************************************AbstractZookeeperClient.java*******************************************************/
//Constants.DYNAMIC_KEY是否创建动态节点，默认为true
public void create(String path, boolean ephemeral) {if (!ephemeral) {// 如果要创建的节点类型非临时节点，那么这里要检测节点是否存在if(persistentExistNodePath.contains(path)){return;}if (checkExists(path)) {persistentExistNodePath.add(path);return;}}int i = path.lastIndexOf('/');if (i > 0) {// 递归创建上一级路径create(path.substring(0, i), false);}// 根据 ephemeral 的值创建临时或持久节点if (ephemeral) {createEphemeral(path);} else {createPersistent(path);persistentExistNodePath.add(path);}}

5、服务发现

5.1 服务发现原理

在官方《Dubbo 用户指南》集群容错部分，给出了服务引用的各功能组件关系图

各节点关系：

这里的 Invoker 是 Provider 的一个可调用 Service 的抽象，Invoker 封装了 Provider 地址及 Service 接口信息
Directory 代表多个 Invoker，可以把它看成 List<Invoker> ，但与 List 不同的是，它的值可能是动态变化的，比如注册中心推送变更
Cluster 将 Directory 中的多个 Invoker 伪装成一个 Invoker，对上层透明，伪装过程包含了容错逻辑，调用失败后，重试另一个
Router 负责从多个 Invoker 中按路由规则选出子集，比如读写分离，应用隔离等
LoadBalance 负责从多个 Invoker 中选出具体的一个用于本次调用，选的过程包含了负载均衡算法，调用失败后，需要重选

Dubbo 服务引用的时机有两个：

1、 Spring 容器调用 ReferenceBean 的 afterPropertiesSet 方法时引用服务

2、 ReferenceBean 对应的服务被注入到其他类中时引用

这两个引用服务的时机区别在于，第一个是饿汉式的，第二个是懒汉式的，默认情况下，Dubbo服务注入使用懒汉式引用服务；如果需要使用饿汉式，可通过配置 <dubbo:reference> 的 init 属性为true时开启

我们以默认的懒汉式引用服务分析：

1、当我们的服务被注入到其他类中时，Spring 会第一时间调用 getObject 方法，并由该方法执行服务引用逻辑

2、通过注册中心引用远程服务，得到一个 Invoker 实例。如果有多个注册中心，多个服务提供者，这个时候会得到一组 Invoker 实例

3、如果得到一组Invoker实例，需要通过集群管理类 Cluster 将多个 Invoker 合并成一个实例

4、合并后的 Invoker 实例已经具备调用本地或远程服务的能力了，但并不能将此实例暴露给用户使用，这会对用户业务代码造成侵入；此时框架还需要通过代理工厂类 (ProxyFactory) 为服务接口生成代理类，并让代理类去调用 Invoker 逻辑

5.2 服务引用入口

服务引用的入口方法为 ReferenceBean 的 getObject 方法，该方法定义在 Spring 的 FactoryBean 接口中，ReferenceBean 实现了这个方法

@Override
public Object getObject() {return get();
}
/*****************************************调用父类的get()方法**********************************************/
public synchronized T get() {if (destroyed) {throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");}if (ref == null) {//肯定是对ref干了些啥，跟进去init();}return ref;}

5.3 创建代理

Dubbo 基于 Spring 的 Schema 扩展实现 XML 配置解析，DubboNamespaceHandler 会将 dubbo:reference 标签解析为 ReferenceBean，ReferenceBean 实现了 FactoryBean，因此当它在代码中有引用时，会调用 ReferenceBean#getObject() 方法进入节点注册和服务发现流程。

/*** ReferenceBean* 服务接口的引用代理，使用transient，引用代理对象实例序列化时不会保存，提高效率*  volatile 变量内存可见，虚拟机不能随意变动优化目标指令，适用一个线程写，多个线程读(只能确定一个线程修改数据之后，其他线程能够看到这个改的。但是两个线程同时修改数据就会产生冲突，也就是脏读)。对比联想单例模式*/
private transient volatile T ref;

/**************************************************ReferenceConfig.java*************************************/
public synchronized T get() {if (destroyed) {throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");}if (ref == null) {init();}return ref;}

private void init() {//.......忽略ref = createProxy(map);
}


private T createProxy(Map<String, String> map) {//.....忽略//服务发现，获取服务响应invokerinvoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));//.....忽略// 创建服务代理，这里一直跟进去可以看到JavassistProxyFactory或者JdkProxyFactoryreturn (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
}


public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {url = getRegistryUrl(url);//获取注册中心对象Registry registry = registryFactory.getRegistry(url);if (RegistryService.class.equals(type)) {return proxyFactory.getInvoker((T) registry, type, url);}
// group="a,b" or group="*"Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));String group = qs.get(GROUP_KEY);//是否有分组if (group != null && group.length() > 0) {if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {return doRefer(Cluster.getCluster(MergeableCluster.NAME), registry, type, url);}}
Cluster cluster = Cluster.getCluster(qs.get(CLUSTER_KEY));//继续跟进return doRefer(cluster, registry, type, url);
}

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);directory.setRegistry(registry);directory.setProtocol(protocol);// all attributes of REFER_KEYMap<String, String> parameters = new HashMap<String, String>(directory.getConsumerUrl().getParameters());//注册中心URL对象URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);if (directory.isShouldRegister()) {directory.setRegisteredConsumerUrl(subscribeUrl);registry.register(directory.getRegisteredConsumerUrl());}directory.buildRouterChain(subscribeUrl);//向 Zookeeper 订阅服务节点信息并 watch 变更，这样就实现了服务自动发现directory.subscribe(toSubscribeUrl(subscribeUrl));//得到invokerInvoker<T> invoker = cluster.join(directory);List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);if (CollectionUtils.isEmpty(listeners)) {return invoker;}
RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker);for (RegistryProtocolListener listener : listeners) {listener.onRefer(this, registryInvokerWrapper);}//进行包装返回return registryInvokerWrapper;}

根据invoker创建代理对象

/*** JavassistRpcProxyFactory*/
public class JavassistProxyFactory extends AbstractProxyFactory {
@Override@SuppressWarnings("unchecked")//生成代理类public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));}
@Overridepublic <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {// TODO Wrapper cannot handle this scenario correctly: the classname contains '$'// 为目标类创建 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);}};}

}

生成的代理类字节文件，反编译过来如下：

/*** Arthas 反编译步骤：* 1. 启动 Arthas*    java -jar arthas-boot.jar** 2. 输入编号选择进程*    Arthas 启动后，会打印 Java 应用进程列表，如下：*    [1]: 11232 org.jetbrains.jps.cmdline.Launcher*    [2]: 22370 org.jetbrains.jps.cmdline.Launcher*    [3]: 22371 com.alibaba.dubbo.demo.consumer.Consumer*    [4]: 22362 com.alibaba.dubbo.demo.provider.Provider*    [5]: 2074 org.apache.zookeeper.server.quorum.QuorumPeerMain* 这里输入编号 3，让 Arthas 关联到启动类为 com.....Consumer 的 Java 进程上** 3. 由于 Demo 项目中只有一个服务接口，因此此接口的代理类类名为 proxy0，此时使用 sc 命令搜索这个类名。*    $ sc *.proxy0*    com.alibaba.dubbo.common.bytecode.proxy0** 4. 使用 jad 命令反编译 com.alibaba.dubbo.common.bytecode.proxy0*    $ jad com.alibaba.dubbo.common.bytecode.proxy0** 更多使用方法请参考 Arthas 官方文档：*   https://alibaba.github.io/arthas/quick-start.html*/
public class proxy0 implements ClassGenerator.DC, EchoService, DemoService {// 方法数组public static Method[] methods;private InvocationHandler handler;
public proxy0(InvocationHandler invocationHandler) {this.handler = invocationHandler;}
public proxy0() {}
public String sayHello(String string) {// 将参数存储到 Object 数组中Object[] arrobject = new Object[]{string};// 调用 InvocationHandler 实现类的 invoke 方法得到调用结果Object object = this.handler.invoke(this, methods[0], arrobject);// 返回调用结果return (String)object;}
/** 回声测试方法 */public Object $echo(Object object) {Object[] arrobject = new Object[]{object};Object object2 = this.handler.invoke(this, methods[1], arrobject);return object2;}
}

5.4 代理调用

当我们调用sayHello时会触发InvokerInvocationHandler .invoke方法

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {if (method.getDeclaringClass() == Object.class) {return method.invoke(invoker, args);}String methodName = method.getName();Class<?>[] parameterTypes = method.getParameterTypes();if (parameterTypes.length == 0) {if ("toString".equals(methodName)) {return invoker.toString();} else if ("$destroy".equals(methodName)) {invoker.destroy();return null;} else if ("hashCode".equals(methodName)) {return invoker.hashCode();}} else if (parameterTypes.length == 1 && "equals".equals(methodName)) {return invoker.equals(args[0]);}//当然我们这个地方不是简单的反射触发调用，而是RPC远程调用RpcInvocation rpcInvocation = new RpcInvocation(method, invoker.getInterface().getName(), args);String serviceKey = invoker.getUrl().getServiceKey();rpcInvocation.setTargetServiceUniqueName(serviceKey);if (consumerModel != null) {rpcInvocation.put(Constants.CONSUMER_MODEL, consumerModel);rpcInvocation.put(Constants.METHOD_MODEL, consumerModel.getMethodModel(method));}//调用return invoker.invoke(rpcInvocation).recreate();
}

进入invoker子类实现：集群模式调用

public class MockClusterInvoker<T> implements Invoker<T> {private final Invoker<T> invoker;public Result invoke(Invocation invocation) throws RpcException {Result result = null;
// 获取 mock 配置值String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();if (value.length() == 0 || value.equalsIgnoreCase("false")) {// 无 mock 逻辑，直接调用其他 Invoker 对象的 invoke 方法，// 比如 FailoverClusterInvokerresult = this.invoker.invoke(invocation);} else if (value.startsWith("force")) {// force:xxx 直接执行 mock 逻辑，不发起远程调用result = doMockInvoke(invocation, null);} else {// fail:xxx 表示消费方对调用服务失败后，再执行 mock 逻辑，不抛出异常try {// 调用其他 Invoker 对象的 invoke 方法result = this.invoker.invoke(invocation);} catch (RpcException e) {if (e.isBiz()) {throw e;} else {// 调用失败，执行 mock 逻辑（服务降级逻辑）result = doMockInvoke(invocation, e);}}}return result;}// 省略其他方法
}

我们找到默认的集群模式invoker：FailoverClusterInvoker

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {List<Invoker<T>> copyInvokers = invokers;checkInvokers(copyInvokers, invocation);String methodName = RpcUtils.getMethodName(invocation);int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;if (len <= 0) {len = 1;}// retry loop.RpcException le = null; // last exception.List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyInvokers.size()); // invoked invokers.Set<String> providers = new HashSet<String>(len);for (int i = 0; i < len; i++) {//Reselect before retry to avoid a change of candidate `invokers`.//NOTE: if `invokers` changed, then `invoked` also lose accuracy.if (i > 0) {checkWhetherDestroyed();copyInvokers = list(invocation);// check againcheckInvokers(copyInvokers, invocation);}//Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);invoked.add(invoker);RpcContext.getContext().setInvokers((List) invoked);try {Result result = invoker.invoke(invocation);//省略
}

处理了一番后最终调用DubboInvoker完成netty client的任务

@Override
protected Result doInvoke(final Invocation invocation) throws Throwable {RpcInvocation inv = (RpcInvocation) invocation;final String methodName = RpcUtils.getMethodName(invocation);inv.setAttachment(PATH_KEY, getUrl().getPath());inv.setAttachment(VERSION_KEY, version);
ExchangeClient currentClient;if (clients.length == 1) {currentClient = clients[0];} else {currentClient = clients[index.getAndIncrement() % clients.length];}try {boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);int timeout = calculateTimeout(invocation, methodName);if (isOneway) {boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);//开始netty client的发送请求currentClient.send(inv, isSent);return AsyncRpcResult.newDefaultAsyncResult(invocation);//省略
}

//-------------------------------NettyChannel.java------------------------------------

public void send(Object message, boolean sent) throws RemotingException {// whether the channel is closedsuper.send(message, sent);
boolean success = true;int timeout = 0;try {ChannelFuture future = channel.writeAndFlush(message);if (sent) {// wait timeout mstimeout = getUrl().getPositiveParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);success = future.await(timeout);}//省略
}