Dubbo（二）：深入理解Dubbo的服务发现SPI机制

时间：2024-02-01 19:37:29

一、前言

　　用到微服务就不得不来谈谈服务发现的话题。通俗的来说，就是在提供服务方把服务注册到注册中心，并且告诉服务消费方现在已经存在了这个服务。那么里面的细节到底是怎么通过代码实现的呢，现在我们来看看Dubbo中的SPI机制

二、SPI简介

　　SPI 全称为 Service Provider Interface，是一种服务发现机制。SPI本质是将接口实现类的全限定名配置在文件中，并由服务加载器读取配置文件，加载实现类，这样运行时可以动态的为接口替换实现类

三、Dubbo中的SPI

　　Dubbo与上面的普通的Java方式实现SPI不同，在Dubbo中重新实现了一套功能更强的SPI机制，即通过键值对的方式进行配置及缓存。其中也使用ConcurrentHashMap与synchronize防止并发问题出现。主要逻辑封装在ExtensionLoader中。下面我们看看源码。

四、ExtensionLoader源码解析

　　由于内部的方法实在太多，我们只挑选与实现SPI的重要逻辑部分拿出来讲解。　　

　　1、getExtensionLoader(Class<T> type)

 1 public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type) {
 2         if (type == null) {
 3             throw new IllegalArgumentException("Extension type == null");
 4         } else if (!type.isInterface()) {
 5             throw new IllegalArgumentException("Extension type (" + type + ") is not an interface!");
 6         } else if (!withExtensionAnnotation(type)) {
 7             throw new IllegalArgumentException("Extension type (" + type + ") is not an extension, because it is NOT annotated with @" + SPI.class.getSimpleName() + "!");
 8         } else {
 9             ExtensionLoader<T> loader = (ExtensionLoader)EXTENSION_LOADERS.get(type);
10             if (loader == null) {
11                 EXTENSION_LOADERS.putIfAbsent(type, new ExtensionLoader(type));
12                 loader = (ExtensionLoader)EXTENSION_LOADERS.get(type);
13             }
14 
15             return loader;
16         }
17     }
View Code

　　这个是可以将对应的接口转换为ExtensionLoader 实例。相当于告诉Dubbo这是个服务接口，里面有对应的服务提供者

　　先是逻辑判断传进来的类不能为空，必须是接口且被@SPI注解注释过。这三个条件都满足就会创建ExtensionLoader 实例。同样的，如果当前类已经被创建过ExtensionLoader 实例，那么直接拿取。否则新建一个。这里使用的是键值对的存储类型，如下图：

 

 

 　　使用ConcurrentHashMap防止在并发时出现问题，并且效率高HashTable不少，所以我们日常项目并发场景中也应该多用ConcurrentHashMap进行存储。

 

　　2、getExtension(String name)

 1 public T getExtension(String name) {
 2     if (name == null || name.length() == 0)
 3         throw new IllegalArgumentException("Extension name == null");
 4     if ("true".equals(name)) {
 5         // 获取默认的拓展实现类
 6         return getDefaultExtension();
 7     }
 8     // Holder，顾名思义，用于持有目标对象
 9     Holder<Object> holder = cachedInstances.get(name);
10     if (holder == null) {
11         cachedInstances.putIfAbsent(name, new Holder<Object>());
12         holder = cachedInstances.get(name);
13     }
14     Object instance = holder.get();
15     // 双重检查
16     if (instance == null) {
17         synchronized (holder) {
18             instance = holder.get();
19             if (instance == null) {
20                 // 创建拓展实例
21                 instance = createExtension(name);
22                 // 设置实例到 holder 中
23                 holder.set(instance);
24             }
25         }
26     }
27     return  instance;
28 }
View Code

　　这个方法主要是相当于得到具体的服务，上述我们已经对服务的接口进行加载，现在我们需要调用服务接口下的某一个具体服务实现类。就用这个方法。上述方法可以看出是会进入getOrCreateHolder中，这个方法顾名思义是获取或者创建Holder。进入到下面方法中：

 1 private Holder<Object> getOrCreateHolder(String name) {
 2         //检查缓存中是否存在
 3         Holder<Object> holder = (Holder)this.cachedInstances.get(name);
 4         if (holder == null) {
 5         //缓存中不存在就去创建一个新的Holder
 6             this.cachedInstances.putIfAbsent(name, new Holder());
 7             holder = (Holder)this.cachedInstances.get(name);
 8         }
 9 
10         return holder;
11     }
View Code

　　同样，缓存池也是以ConcurrentHashMap为存储结构

 

 

　　 3、createExtension(String name)

　　实际上getExtension方法不一定每次都能拿到，当服务实现类是第一次进行加载的时候就需要当前的方法

 1 private T createExtension(String name) {
 2         Class<?> clazz = (Class)this.getExtensionClasses().get(name);
 3         if (clazz == null) {
 4             throw this.findException(name);
 5         } else {
 6             try {
 7                 T instance = EXTENSION_INSTANCES.get(clazz);
 8                 if (instance == null) {
 9                     EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.newInstance());
10                     instance = EXTENSION_INSTANCES.get(clazz);
11                 }
12 
13                 this.injectExtension(instance);
14                 Set<Class<?>> wrapperClasses = this.cachedWrapperClasses;
15                 Class wrapperClass;
16                 if (CollectionUtils.isNotEmpty(wrapperClasses)) {
17                     for(Iterator var5 = wrapperClasses.iterator(); var5.hasNext(); instance = this.injectExtension(wrapperClass.getConstructor(this.type).newInstance(instance))) {
18                         wrapperClass = (Class)var5.next();
19                     }
20                 }
21 
22                 return instance;
23             } catch (Throwable var7) {
24                 throw new IllegalStateException("Extension instance (name: " + name + ", class: " + this.type + ") couldn't be instantiated: " + var7.getMessage(), var7);
25             }
26         }
27     }
View Code

　　可以看出createExtension实际上是一个私有方法，也就是由上面的getExtension自动触发。内部逻辑大致为：

　　　　3.1、通过 getExtensionClasses 获取所有的拓展类

　　　　3.2、通过反射创建拓展对象

　　　　3.3、向拓展对象中注入依赖（这里Dubbo有单独的IOC后面会介绍）

　　　　3.4、将拓展对象包裹在相应的 Wrapper 对象中

　　4、getExtensionClasses()

  1 private Map<String, Class<?>> getExtensionClasses() {
  2     // 从缓存中获取已加载的拓展类
  3     Map<String, Class<?>> classes = cachedClasses.get();
  4     // 双重检查
  5     if (classes == null) {
  6         synchronized (cachedClasses) {
  7             classes = cachedClasses.get();
  8             if (classes == null) {
  9                 // 加载拓展类
 10                 classes = loadExtensionClasses();
 11                 cachedClasses.set(classes);
 12             }
 13         }
 14     }
 15     return classes;
 16 }
 17 
 18 //进入到loadExtensionClasses中
 19 
 20 private Map<String, Class<?>> loadExtensionClasses() {
 21     // 获取 SPI 注解，这里的 type 变量是在调用 getExtensionLoader 方法时传入的
 22     final SPI defaultAnnotation = type.getAnnotation(SPI.class);
 23     if (defaultAnnotation != null) {
 24         String value = defaultAnnotation.value();
 25         if ((value = value.trim()).length() > 0) {
 26             // 对 SPI 注解内容进行切分
 27             String[] names = NAME_SEPARATOR.split(value);
 28             // 检测 SPI 注解内容是否合法，不合法则抛出异常
 29             if (names.length > 1) {
 30                 throw new IllegalStateException("more than 1 default extension name on extension...");
 31             }
 32 
 33             // 设置默认名称，参考 getDefaultExtension 方法
 34             if (names.length == 1) {
 35                 cachedDefaultName = names[0];
 36             }
 37         }
 38     }
 39 
 40     Map<String, Class<?>> extensionClasses = new HashMap<String, Class<?>>();
 41     // 加载指定文件夹下的配置文件
 42     loadDirectory(extensionClasses, DUBBO_INTERNAL_DIRECTORY);
 43     loadDirectory(extensionClasses, DUBBO_DIRECTORY);
 44     loadDirectory(extensionClasses, SERVICES_DIRECTORY);
 45     return extensionClasses;
 46 }
 47 
 48 //进入到loadDirectory中
 49 
 50 private void loadDirectory(Map<String, Class<?>> extensionClasses, String dir) {
 51     // fileName = 文件夹路径 + type 全限定名 
 52     String fileName = dir + type.getName();
 53     try {
 54         Enumeration<java.net.URL> urls;
 55         ClassLoader classLoader = findClassLoader();
 56         // 根据文件名加载所有的同名文件
 57         if (classLoader != null) {
 58             urls = classLoader.getResources(fileName);
 59         } else {
 60             urls = ClassLoader.getSystemResources(fileName);
 61         }
 62         if (urls != null) {
 63             while (urls.hasMoreElements()) {
 64                 java.net.URL resourceURL = urls.nextElement();
 65                 // 加载资源
 66                 loadResource(extensionClasses, classLoader, resourceURL);
 67             }
 68         }
 69     } catch (Throwable t) {
 70         logger.error("...");
 71     }
 72 }
 73 
 74 //进入到loadResource中
 75 
 76 private void loadResource(Map<String, Class<?>> extensionClasses, 
 77     ClassLoader classLoader, java.net.URL resourceURL) {
 78     try {
 79         BufferedReader reader = new BufferedReader(
 80             new InputStreamReader(resourceURL.openStream(), "utf-8"));
 81         try {
 82             String line;
 83             // 按行读取配置内容
 84             while ((line = reader.readLine()) != null) {
 85                 // 定位 # 字符
 86                 final int ci = line.indexOf('#');
 87                 if (ci >= 0) {
 88                     // 截取 # 之前的字符串，# 之后的内容为注释，需要忽略
 89                     line = line.substring(0, ci);
 90                 }
 91                 line = line.trim();
 92                 if (line.length() > 0) {
 93                     try {
 94                         String name = null;
 95                         int i = line.indexOf('=');
 96                         if (i > 0) {
 97                             // 以等于号 = 为界，截取键与值
 98                             name = line.substring(0, i).trim();
 99                             line = line.substring(i + 1).trim();
100                         }
101                         if (line.length() > 0) {
102                             // 加载类，并通过 loadClass 方法对类进行缓存
103                             loadClass(extensionClasses, resourceURL, 
104                                       Class.forName(line, true, classLoader), name);
105                         }
106                     } catch (Throwable t) {
107                         IllegalStateException e = new IllegalStateException("Failed to load extension class...");
108                     }
109                 }
110             }
111         } finally {
112             reader.close();
113         }
114     } catch (Throwable t) {
115         logger.error("Exception when load extension class...");
116     }
117 }
118 
119 //进入到loadClass中
120 
121 private void loadClass(Map<String, Class<?>> extensionClasses, java.net.URL resourceURL, 
122     Class<?> clazz, String name) throws NoSuchMethodException {
123     
124     if (!type.isAssignableFrom(clazz)) {
125         throw new IllegalStateException("...");
126     }
127 
128     // 检测目标类上是否有 Adaptive 注解
129     if (clazz.isAnnotationPresent(Adaptive.class)) {
130         if (cachedAdaptiveClass == null) {
131             // 设置 cachedAdaptiveClass缓存
132             cachedAdaptiveClass = clazz;
133         } else if (!cachedAdaptiveClass.equals(clazz)) {
134             throw new IllegalStateException("...");
135         }
136         
137     // 检测 clazz 是否是 Wrapper 类型
138     } else if (isWrapperClass(clazz)) {
139         Set<Class<?>> wrappers = cachedWrapperClasses;
140         if (wrappers == null) {
141             cachedWrapperClasses = new ConcurrentHashSet<Class<?>>();
142             wrappers = cachedWrapperClasses;
143         }
144         // 存储 clazz 到 cachedWrapperClasses 缓存中
145         wrappers.add(clazz);
146         
147     // 程序进入此分支，表明 clazz 是一个普通的拓展类
148     } else {
149         // 检测 clazz 是否有默认的构造方法，如果没有，则抛出异常
150         clazz.getConstructor();
151         if (name == null || name.length() == 0) {
152             // 如果 name 为空，则尝试从 Extension 注解中获取 name，或使用小写的类名作为 name
153             name = findAnnotationName(clazz);
154             if (name.length() == 0) {
155                 throw new IllegalStateException("...");
156             }
157         }
158         // 切分 name
159         String[] names = NAME_SEPARATOR.split(name);
160         if (names != null && names.length > 0) {
161             Activate activate = clazz.getAnnotation(Activate.class);
162             if (activate != null) {
163                 // 如果类上有 Activate 注解，则使用 names 数组的第一个元素作为键，
164                 // 存储 name 到 Activate 注解对象的映射关系
165                 cachedActivates.put(names[0], activate);
166             }
167             for (String n : names) {
168                 if (!cachedNames.containsKey(clazz)) {
169                     // 存储 Class 到名称的映射关系
170                     cachedNames.put(clazz, n);
171                 }
172                 Class<?> c = extensionClasses.get(n);
173                 if (c == null) {
174                     // 存储名称到 Class 的映射关系
175                     extensionClasses.put(n, clazz);
176                 } else if (c != clazz) {
177                     throw new IllegalStateException("...");
178                 }
179             }
180         }
181     }
182 }
View Code

　　上面的方法较多，理一下逻辑：

　　1、getExtensionClasses()：先检查缓存，若缓存未命中，则通过 synchronized 加锁。加锁后再次检查缓存，并判断是否为空。此时如果 classes 仍为 null，则通过 loadExtensionClasses 加载拓展类。

　　2、loadExtensionClasses()：对 SPI 注解的接口进行解析，而后调用 loadDirectory 方法加载指定文件夹配置文件。

　　3、loadDirectory()：方法先通过 classLoader 获取所有资源链接，然后再通过 loadResource 方法加载资源。

　　4、loadResource()：用于读取和解析配置文件，并通过反射加载类，最后调用 loadClass 方法进行其他操作。loadClass 方法用于主要用于操作缓存。

　　5、小结：

　　我们稍微捋一下Dubbo是如何进行SPI的即发现接口的实现类。先是需要实例化扩展类加载器。这里为了更好的和微服务贴合起来，我们就把它称作服务加载器。在服务加载器中用的是ConcurrentHashMap的缓存结构。在我们需要寻找服务的过程中，Dubbo先通过反射加载类，而后将有@SPI表示的接口（即服务接口）的实现类（即服务提供方）进行配置对应的文件夹及文件。将配置文件以键值对的方式存到缓存中key就是当前服务接口下类的名字，value就是Dubbo生成的对应的类配置文件。方便我们下次调用。其中为了防止并发问题产生，使用ConcurrentHashMap，并且使用synchronize关键字对存在并发问题的节点进行双重检查。

五、Dubbo中的IOC

　　在createExtension中有提到过将拓展对象注入依赖。这里使用的是injectExtension(T instance)：

 1 private T injectExtension(T instance) {
 2     try {
 3         if (objectFactory != null) {
 4             // 遍历目标类的所有方法
 5             for (Method method : instance.getClass().getMethods()) {
 6                 // 检测方法是否以 set 开头，且方法仅有一个参数，且方法访问级别为 public
 7                 if (method.getName().startsWith("set")
 8                     && method.getParameterTypes().length == 1
 9                     && Modifier.isPublic(method.getModifiers())) {
10                     // 获取 setter 方法参数类型
11                     Class<?> pt = method.getParameterTypes()[0];
12                     try {
13                         // 获取属性名，比如 setName 方法对应属性名 name
14                         String property = method.getName().length() > 3 ? 
15                             method.getName().substring(3, 4).toLowerCase() + 
16                                 method.getName().substring(4) : "";
17                         // 从 ObjectFactory 中获取依赖对象
18                         Object object = objectFactory.getExtension(pt, property);
19                         if (object != null) {
20                             // 通过反射调用 setter 方法设置依赖
21                             method.invoke(instance, object);
22                         }
23                     } catch (Exception e) {
24                         logger.error("fail to inject via method...");
25                     }
26                 }
27             }
28         }
29     } catch (Exception e) {
30         logger.error(e.getMessage(), e);
31     }
32     return instance;
33 }
View Code

　　在上面代码中，objectFactory 变量的类型为 AdaptiveExtensionFactory，AdaptiveExtensionFactory 内部维护了一个 ExtensionFactory 列表，用于存储其他类型的 ExtensionFactory。Dubbo 目前提供了两种 ExtensionFactory，分别是 SpiExtensionFactory 和 SpringExtensionFactory。前者用于创建自适应的拓展，后者是用于从 Spring 的 IOC 容器中获取所需的拓展。这就是我们常说的Dubbo为什么能够与Spring无缝连接，因为Dubbo底层就是依赖Spring的，对于Spring的IOC容器可直接拿来用。

六、总结

　　从框架的源码中如果要继续深挖的话，可以多思考思考synchronize用的地方，为什么要用，如果不用的话会有什么并发问题。Dubbo的服务发现只是为我们以后学习Dubbo框架打下基础，至少让我们知道Dubbo是如何进行服务发现的。

 

 

 

 

　　

　　

相关文章