前言
前面我們研究了RPC的原理，市面上有很多基于RPC思想實作的架構，比如有Dubbo。今天就從Dubbo的SPI機制、服務注冊與發現源碼及網絡通信過程去深入剖析下Dubbo。
Dubbo架構
概述
Dubbo是阿裡巴巴公司開源的一個高性能優秀的服務架構，使得應用可通過高性能的RPC 實作服務的輸出和輸入功能，可以和Spring架構無縫內建。
Dubbo是一款高性能、輕量級的開源Java RPC架構，它提供了三大核心能力：面向接口的遠端方法調用，智能容錯和負載均衡，以及服務自動注冊和發現。
調用流程:
- 服務容器負責啟動，加載，運作服務提供者。
- 服務提供者在啟動時，向注冊中心注冊自己提供的服務。
- 服務消費者在啟動時，向注冊中心訂閱自己所需的服務。
- 注冊中心傳回服務提供者位址清單給消費者，如果有變更，注冊中心将基于長連接配接推送變更資料給消費者。
- 服務消費者，從提供者位址清單中，基于軟負載均衡算法，選一台提供者進行調用，如果調用失敗，再選另一台調用。
- 服務消費者和提供者，在記憶體中累計調用次數和調用時間，定時每分鐘發送一次統計資料到監控中心。
架構體系
源碼結構
- dubbo-common:公共邏輯子產品: 包括Util類和通用模型
- dubbo-remoting 遠端通信子產品: 相當于dubbo協定的實作，如果RPC使用RMI協定則不需要使用此包
- dubbo-rpc 遠端調用子產品: 抽象各種協定，以及動态代理，包含一對一的調用，不關心叢集的原理。
- dubbo-cluster 叢集子產品: 将多個服務提供方僞裝成一個提供方,包括負載均衡,容錯,路由等,叢集的位址清單可以是靜态配置的,也可以是注冊中心下發的.
- dubbo-registry 注冊中心子產品: 基于注冊中心下發的叢集方式,以及對各種注冊中心的抽象
- dubbo-monitor 監控子產品: 統計服務調用次數,調用時間,調用鍊跟蹤的服務.
- dubbo-config 配置子產品: 是dubbo對外的api,使用者通過config使用dubbo,隐藏dubbo所有細節
- dubbo-container 容器子產品: 是一個standlone的容器,以簡單的main加載spring啟動,因為服務通常不需要Tomcat/Jboss等web容器的特性,沒必要用web容器去加載服務.
整體設計
- 圖中左邊淡藍背景的為服務消費方使用的接口，右邊淡綠色背景的為服務提供方使用的接口，位于中軸線上的為雙方都用到的接口。
- 圖中從下至上分為十層，各層均為單向依賴，每一層都可以剝離上層被複用，其中，Service 和Config 層為API，其它各層均為SPI。
- 圖中綠色小塊的為擴充接口，藍色小塊為實作類，圖中隻顯示用于關聯各層的實作類。
- 圖中藍色虛線為初始化過程，即啟動時組裝鍊，紅色實線為方法調用過程，即運作時調時鍊，紫色三角箭頭為繼承，可以把子類看作父類的同一個節點，線上的文字為調用的方法。
各層說明
- config 配置層：對外配置接口，以 ServiceConfig , ReferenceConfig 為中心，可以直接初始化配置類，也可以通過spring 解析配置生成配置類
- proxy 服務代理層：服務接口透明代理，生成服務的用戶端Stub 和伺服器端Skeleton, 以ServiceProxy 為中心，擴充接口為 ProxyFactory
- registry 注冊中心層：封裝服務位址的注冊與發現，以服務URL 為中心，擴充接口為RegistryFactory , Registry , RegistryService
- cluster 路由層：封裝多個提供者的路由及負載均衡，并橋接注冊中心，以 Invoker 為中心，擴充接口為 Cluster , Directory , Router , LoadBalance
- monitor 監控層：RPC 調用次數和調用時間監控，以 Statistics 為中心，擴充接口為MonitorFactory , Monitor , MonitorService
- protocol 遠端調用層：封裝RPC 調用，以 Invocation , Result 為中心，擴充接口為Protocol , Invoker , Exporter
- exchange 資訊交換層：封裝請求響應模式，同步轉異步，以 Request , Response 為中心，擴充接口為 Exchanger , ExchangeChannel , ExchangeClient , ExchangeServer
- transport 網絡傳輸層：抽象mina 和netty 為統一接口，以 Message 為中心，擴充接口為Channel , Transporter , Client , Server , Codec
- serialize 資料序列化層：可複用的一些工具，擴充接口為 Serialization , ObjectInput ,ObjectOutput , ThreadPool
調用流程
對照上面的整體架構圖可以大緻分為以下步驟：
1、服務提供者啟動，開啟Netty服務，建立Zookeeper用戶端，向注冊中心注冊服務。
2、服務消費者啟動，通過Zookeeper向注冊中心擷取服務提供者清單，與服務提供者通過Netty建立長連接配接。
3、服務消費者通過接口開始遠端調用服務，ProxyFactory通過初始化Proxy對象，Proxy通過建立動态代理對象。
4、動态代理對象通過invoke方法，層層包裝生成一個Invoker對象，該對象包含了代理對象。
5、Invoker通過路由，負載均衡選擇了一個最合适的服務提供者，在通過加入各種過濾器，協定層包裝生成一個新的DubboInvoker對象。
6、再通過交換成将DubboInvoker對象包裝成一個Reuqest對象，該對象通過序列化通過NettyClient傳輸到服務提供者的NettyServer端。
7、到了服務提供者這邊，再通過反序列化、協定解密等操作生成一個DubboExporter對象,再層層傳遞處理,會生成一個服務提供端的Invoker對象.
8、這個Invoker對象會調用本地服務，獲得結果再通過層層回調傳回到服務消費者，服務消費者拿到結果後，再解析獲得最終結果。
Dubbo中的SPI機制
什麼是SPI
在Dubbo 中，SPI 是一個非常重要的子產品。基于SPI，我們可以很容易的對Dubbo 進行拓展。如果大家想要學習Dubbo 的源碼，SPI 機制務必弄懂。接下來，我們先來了解一下Java SPI 與Dubbo SPI 的用法，然後再來分析Dubbo SPI 的源碼。
SPI是Service Provider Interface 服務提供接口縮寫,是一種服務發現機制。SPI的本質是将接口的實作類的全限定名定義在配置檔案中，并有伺服器讀取配置檔案，并加載實作類。這樣就可以在運作的時候，動态為接口替換實作類。
JDK中的SPI
Java SPI 實際上是“基于接口的程式設計＋政策模式＋配置檔案”組合實作的動态加載機制。
通過一個案例我們來認識下SPI
定義一個接口：
package com.laowang;
/**
* @author 原
* @date 2021/3/27
* @since 1.0
**/
public interface User {
String showName();
} 定義兩個實作類
package com.laowang.impl;
import com.laowang.User;
/**
* @author 原
* @date 2021/3/27
* @since 1.0
**/
public class Student implements User {
@Override
public String showName() {
System.out.println("my name is laowang");
return null;
}
} package com.laowang.impl;
import com.laowang.User;
/**
* @author 原
* @date 2021/3/27
* @since 1.0
**/
public class Teacher implements User {
@Override
public String showName() {
System.out.println("my name is zhangsan");
return null;
}
} 在resources目錄下建立檔案夾META-INF.services，并在該檔案夾下建立一個名稱與User的全路徑一緻的檔案com.laowang.User
在檔案中寫入，兩個實作類的全路徑名
編寫測試類：
package com.laowang;
import java.util.ServiceLoader;
/**
* @author 原
* @date 2021/3/27
* @since 1.0
**/
public class SpiTest {
public static void main(String[] args) {
ServiceLoader<User> serviceLoader = ServiceLoader.load(User.class);
serviceLoader.forEach(User::showName);
}
} 運作結果:
我們發現通過SPI機制，幫我們自動運作了兩個實作類。
通過檢視ServiceLoader源碼:
其實通過讀取配置檔案中實作類的全路徑類名,通過反射建立對象,并放入providers容器中。
總結：
調用過程
應用程式調用ServiceLoader.load方法，建立一個新的ServiceLoader，并執行個體化該類中的成員變量
應用程式通過疊代器接口擷取對象執行個體，ServiceLoader先判斷成員變量providers對象中(LinkedHashMap<String,S>類型)是否有緩存執行個體對象，如果有緩存，直接傳回。如果沒有緩存，執行類的裝載，
優點
使用Java SPI 機制的優勢是實作解耦，使得接口的定義與具體業務實作分離，而不是耦合在一起。應用程序可以根據實際業務情況啟用或替換具體元件。
缺點
不能按需加載。雖然ServiceLoader 做了延遲載入，但是基本隻能通過周遊全部擷取，也就是接口的實作類得全部載入并執行個體化一遍。如果你并不想用某些實作類，或者某些類執行個體化很耗時，它也被載入并執行個體化了，這就造成了浪費。
擷取某個實作類的方式不夠靈活，隻能通過Iterator 形式擷取，不能根據某個參數來擷取對應的實作類。
多個并發多線程使用ServiceLoader 類的執行個體是不安全的。
加載不到實作類時抛出并不是真正原因的異常，錯誤很難定位。
Dubbo中的SPI
Dubbo 并未使用Java SPI，而是重新實作了一套功能更強的SPI 機制。Dubbo SPI 的相關邏輯被封裝在了ExtensionLoader 類中，通過ExtensionLoader，我們可以加載指定的實作類。
栗子
與Java SPI 實作類配置不同，Dubbo SPI 是通過鍵值對的方式進行配置，這樣我們可以按需加載指定的實作類。下面來示範Dubbo SPI 的用法：
Dubbo SPI 所需的配置檔案需放置在META-INF/dubbo 路徑下，與Java SPI 實作類配置不同，DubboSPI 是通過鍵值對的方式進行配置，配置内容如下。
optimusPrime = org.apache.spi.OptimusPrime
bumblebee = org.apache.spi.Bumblebee 在使用Dubbo SPI 時，需要在接口上标注@SPI 注解。
@SPI
public interface Robot {
void sayHello();
} 通過ExtensionLoader，我們可以加載指定的實作類，下面來示範Dubbo SPI ：
public class DubboSPITest {
@Test
public void sayHello() throws Exception {
ExtensionLoader<Robot> extensionLoader =
ExtensionLoader.getExtensionLoader(Robot.class);
Robot optimusPrime = extensionLoader.getExtension("optimusPrime");
optimusPrime.sayHello();
Robot bumblebee = extensionLoader.getExtension("bumblebee");
bumblebee.sayHello();
}
} Dubbo SPI 除了支援按需加載接口實作類，還增加了IOC 和AOP 等特性，這些特性将會在接下來的源碼分析章節中一一進行介紹。
源碼分析
ExtensionLoader 的getExtensionLoader 方法擷取一個ExtensionLoader 執行個體，然後再通過ExtensionLoader 的getExtension 方法擷取拓展類對象。下面我們從ExtensionLoader 的getExtension 方法作為入口，對拓展類對象的擷取過程進行詳細的分析。
public T getExtension(String name) {
if (StringUtils.isEmpty(name)) {
throw new IllegalArgumentException("Extension name == null");
}
if ("true".equals(name)) {
// 擷取預設的拓展實作類
return getDefaultExtension();
}
// Holder，顧名思義，用于持有目标對象 就是從容器中擷取,如果沒有直接new一個Holder
Holder<Object> holder = getOrCreateHolder(name);
//擷取目标對象執行個體
Object instance = holder.get();
// 如果目标對象執行個體為null 就需要通過雙重檢查建立執行個體
if (instance == null) {
synchronized (holder) {
instance = holder.get();
if (instance == null) {
// 建立拓展執行個體
instance = createExtension(name);
// 設定執行個體到 holder 中
holder.set(instance);
}
}
}
return (T) instance;
} 上面代碼的邏輯比較簡單，首先檢查緩存，緩存未命中則建立拓展對象。下面我們來看一下建立拓展對象的過程是怎樣的。
private T createExtension(String name) {
// 從配置檔案中加載所有的拓展類，可得到“配置項名稱”到“配置類”的映射關系表
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);
}
// 向執行個體中注入依賴 下面是IOC和AOP的實作
injectExtension(instance);
Set<Class<?>> wrapperClasses = cachedWrapperClasses;
if (CollectionUtils.isNotEmpty(wrapperClasses)) {
// 循環建立 Wrapper 執行個體
for (Class<?> wrapperClass : wrapperClasses) {
// 将目前 instance 作為參數傳給 Wrapper 的構造方法，并通過反射建立Wrapper 執行個體。
// 然後向 Wrapper 執行個體中注入依賴，最後将 Wrapper 執行個體再次指派給instance 變量
instance = injectExtension(
(T)
wrapperClass.getConstructor(type).newInstance(instance));
}
} createExtension 方法的邏輯稍複雜一下，包含了如下的步驟：
- 通過getExtensionClasses 擷取所有的拓展類
- 通過反射建立拓展對象
- 向拓展對象中注入依賴
-
将拓展對象包裹在相應的Wrapper 對象中
以上步驟中，第一個步驟是加載拓展類的關鍵，第三和第四個步驟是Dubbo IOC 與AOP 的具體實作。由于此類設計源碼較多，這裡簡單的總結下ExtensionLoader整個執行邏輯：
getExtension(String name) #根據key擷取拓展對象 -->createExtension(String name) #建立拓展執行個體 -->getExtensionClasses #根據路徑擷取所有的拓展類 -->loadExtensionClasses #加載拓展類 -->cacheDefaultExtensionName #解析@SPI注解 -->loadDirectory #方法加載指定檔案夾配置檔案 -->loadResource #加載資源 -->loadClass #加載類，并通過 loadClass 方法對類進行緩存
Dubbo的SPI如何實作IOC和AOP的
Dubbo IOC
Dubbo IOC 是通過setter 方法注入依賴。Dubbo 首先會通過反射擷取到執行個體的所有方法，然後再周遊方法清單，檢測方法名是否具有setter 方法特征。若有，則通過ObjectFactory 擷取依賴對象，最後通過反射調用setter 方法将依賴設定到目标對象中。整個過程對應的代碼如下：
private T injectExtension(T instance) {
try {
if (objectFactory != null) {
//擷取執行個體的所有方法
for (Method method : instance.getClass().getMethods()) {
//isSetter做的事:檢測方法是否以 set 開頭，且方法僅有一個參數，且方法通路級别為 public
if (isSetter(method)) {
/**
* Check {@link DisableInject} to see if we need auto injection for this property
*/
if (method.getAnnotation(DisableInject.class) != null) {
continue;
}
Class<?> pt = method.getParameterTypes()[0];
if (ReflectUtils.isPrimitives(pt)) {
continue;
}
try {
String property = getSetterProperty(method);
//擷取依賴對象
Object object = objectFactory.getExtension(pt, property);
if (object != null) {
//設定屬性
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;
} Dubbo Aop
在說這個之前，我們得先知道裝飾者模式
裝飾者模式：在不改變原類檔案以及不使用繼承的情況下，動态地将責任附加到對象上，進而實作動态拓展一個對象的功能。它是通過建立一個包裝對象，也就是裝飾來包裹真實的對象。
在用Spring的時候，我們經常會用到AOP功能。在目标類的方法前後插入其他邏輯。比如通常使用Spring AOP來實作日志，監控和鑒權等功能。Dubbo的擴充機制，是否也支援類似的功能呢？答案是yes。在Dubbo中，有一種特殊的類，被稱為Wrapper類。通過裝飾者模式，使用包裝類包裝原始的擴充點執行個體。在原始擴充點實作前後插入其他邏輯，實作AOP功能。
一般來說裝飾者模式有下面幾個參與者：
Component：裝飾者和被裝飾者共同的父類，是一個接口或者抽象類，用來定義基本行為
ConcreteComponent：定義具體對象，即被裝飾者
Decorator：抽象裝飾者，繼承自Component，從外類來擴充ConcreteComponent。對于ConcreteComponent來說，不需要知道Decorator的存在，Decorator是一個接口或抽象類
ConcreteDecorator：具體裝飾者，用于擴充ConcreteComponent
//擷取所有需要包裝的類
Set<Class<?>> wrapperClasses = cachedWrapperClasses; 我們再看看cachedWrapperClasses是什麼?
private Set<Class<?>> cachedWrapperClasses; 是一個set集合,那麼集合是什麼時候添加元素的呢?
/**
* cache wrapper class
* <p>
* like: ProtocolFilterWrapper, ProtocolListenerWrapper
*/
private void cacheWrapperClass(Class<?> clazz) {
if (cachedWrapperClasses == null) {
cachedWrapperClasses = new ConcurrentHashSet<>();
}
cachedWrapperClasses.add(clazz);
} 通過這個方法添加的,再看看誰調用了這個私有方法:
/**
* test if clazz is a wrapper class
* <p>
* which has Constructor with given class type as its only argument
*/
private boolean isWrapperClass(Class<?> clazz) {
try {
clazz.getConstructor(type);
return true;
} catch (NoSuchMethodException e) {
return false;
}
} 原來是通過isWrapperClass這個方法,判斷有沒有其他對象中的構造方法中持有本對象,如果有,dubbo就認為這是個裝飾類,調用裝飾者類的構造方法，并傳回執行個體對象
然後通過執行個體化這個包裝類代替需要加載的這個類。這樣執行的方法就是包裝類的方法。
Dubbo中的動态編譯
我們知道在Dubbo 中，很多拓展都是通過SPI 機制 進行加載的，比如Protocol、Cluster、LoadBalance、ProxyFactory 等。有時，有些拓展并不想在架構啟動階段被加載，而是希望在拓展方法被調用時，根據運作時參數進行加載，即根據參數動态加載實作類。
這種在運作時，根據方法參數才動态決定使用具體的拓展，在dubbo中就叫做擴充點自适應執行個體。其實是一個擴充點的代理，将擴充的選擇從Dubbo啟動時，延遲到RPC調用時。Dubbo中每一個擴充點都有一個自适應類，如果沒有顯式提供，Dubbo會自動為我們建立一個，預設使用Javaassist。
自适應拓展機制的實作邏輯是這樣的
- 首先Dubbo 會為拓展接口生成具有代理功能的代碼；
- 通過javassist 或jdk 編譯這段代碼，得到Class 類；
- 通過反射建立代理類；
- 在代理類中，通過URL對象的參數來确定到底調用哪個實作類；
javassist
Javassist是一個開源的分析、編輯和建立Java位元組碼的類庫。是由東京工業大學的數學和計算機科學系的Shigeru Chiba （千葉滋）所建立的。它已加入了開放源代碼JBoss 應用伺服器項目,通過使用Javassist對位元組碼操作為JBoss實作動态AOP架構。javassist是jboss的一個子項目，其主要的優點，在于簡單，而且快速。直接使用java編碼的形式，而不需要了解虛拟機指令，就能動态改變類的結構，或者動态生成類。
/**
* Javassist是一個開源的分析、編輯和建立Java位元組碼的類庫
* 能動态改變類的結構，或者動态生成類
*/
public class CompilerByJavassist {
public static void main(String[] args) throws Exception {
// ClassPool：class對象容器
ClassPool pool = ClassPool.getDefault();
// 通過ClassPool生成一個User類
CtClass ctClass = pool.makeClass("com.itheima.domain.User");
// 添加屬性 -- private String username
CtField enameField = new CtField(pool.getCtClass("java.lang.String"),
"username", ctClass);
enameField.setModifiers(Modifier.PRIVATE);
ctClass.addField(enameField);
// 添加屬性 -- private int age
CtField enoField = new CtField(pool.getCtClass("int"), "age", ctClass);
enoField.setModifiers(Modifier.PRIVATE);
ctClass.addField(enoField);
//添加方法
ctClass.addMethod(CtNewMethod.getter("getUsername", enameField));
ctClass.addMethod(CtNewMethod.setter("setUsername", enameField));
ctClass.addMethod(CtNewMethod.getter("getAge", enoField));
ctClass.addMethod(CtNewMethod.setter("setAge", enoField));
// 無參構造器
CtConstructor constructor = new CtConstructor(null, ctClass);
constructor.setBody("{}");
ctClass.addConstructor(constructor);
// 添加構造函數
//ctClass.addConstructor(new CtConstructor(new CtClass[] {}, ctClass));
CtConstructor ctConstructor = new CtConstructor(new CtClass[]
{pool.get(String.class.getName()),CtClass.intType}, ctClass);
ctConstructor.setBody("{\n this.username=$1; \n this.age=$2;\n}");
ctClass.addConstructor(ctConstructor);
// 添加自定義方法
CtMethod ctMethod = new CtMethod(CtClass.voidType, "printUser",new
CtClass[] {}, ctClass);
// 為自定義方法設定修飾符
ctMethod.setModifiers(Modifier.PUBLIC);
// 為自定義方法設定函數體
StringBuffer buffer2 = new StringBuffer();
buffer2.append("{\nSystem.out.println(\"使用者資訊如下\");\n")
.append("System.out.println(\"使用者名=\"+username);\n")
.append("System.out.println(\"年齡=\"+age);\n").append("}");
ctMethod.setBody(buffer2.toString());
ctClass.addMethod(ctMethod);
//生成一個class
Class<?> clazz = ctClass.toClass();
Constructor cons2 =
clazz.getDeclaredConstructor(String.class,Integer.TYPE);
Object obj = cons2.newInstance("itheima",20);
//反射 執行方法
obj.getClass().getMethod("printUser", new Class[] {})
.invoke(obj, new Object[] {});
// 把生成的class檔案寫入檔案
byte[] byteArr = ctClass.toBytecode();
FileOutputStream fos = new FileOutputStream(new File("D://User.class"));
fos.write(byteArr);
fos.close();
}
} Adaptive注解
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE, ElementType.METHOD})
public @interface Adaptive {
String[] value() default {};
} Adaptive 可注解在類或方法上。
标注在類上：Dubbo 不會為該類生成代理類。
标注在方法上：Dubbo 則會為該方法生成代理邏輯，表示目前方法需要根據 參數URL 調用對應的擴充點實作。
dubbo中每一個擴充點都有一個自适應類，如果沒有顯式提供，Dubbo會自動為我們建立一個，預設使用Javaassist。 先來看下建立自适應擴充類的代碼
//1、看下extensionLoader的擷取方法
ExtensionLoader<Robot>extensionLoader=ExtensionLoader.getExtensionLoader(Robot.class);
//2、最終調用的是ExtensionLoader的構造方法
private ExtensionLoader(Class<?> type) {
this.type = type;
objectFactory = (type == ExtensionFactory.class ? null : ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getAdaptiveExtension());
}
//3、getAdaptiveExtension()看看幹了什麼事
public T getAdaptiveExtension() {
//擷取自适應擴充類,如果沒有就開始初始化一個
Object instance = cachedAdaptiveInstance.get();
if (instance == null) {
if (createAdaptiveInstanceError == null) {
synchronized (cachedAdaptiveInstance) {
instance = cachedAdaptiveInstance.get();
if (instance == null) {
try {
//這裡建立了一個自适應擴充類
instance = createAdaptiveExtension();
cachedAdaptiveInstance.set(instance);
} catch (Throwable t) {
createAdaptiveInstanceError = t;
throw new IllegalStateException("Failed to create adaptive instance: " + t.toString(), t);
}
}
}
} else {
throw new IllegalStateException("Failed to create adaptive instance: " + createAdaptiveInstanceError.toString(), createAdaptiveInstanceError);
}
}
return (T) instance;
}
//看看createAdaptiveExtension()
private T createAdaptiveExtension() {
try {
return injectExtension((T) getAdaptiveExtensionClass().newInstance());
} catch (Exception e) {
throw new IllegalStateException("Can't create adaptive extension " + type + ", cause: " + e.getMessage(), e);
}
}
//再進到getAdaptiveExtensionClass()
private Class<?> getAdaptiveExtensionClass() {
getExtensionClasses();
if (cachedAdaptiveClass != null) {
return cachedAdaptiveClass;
}
return cachedAdaptiveClass = createAdaptiveExtensionClass();
}
//繼續追進去createAdaptiveExtensionClass()
private Class<?> createAdaptiveExtensionClass() {
String code = new AdaptiveClassCodeGenerator(type, cachedDefaultName).generate();
ClassLoader classLoader = findClassLoader();
org.apache.dubbo.common.compiler.Compiler compiler = ExtensionLoader.getExtensionLoader(org.apache.dubbo.common.compiler.Compiler.class).getAdaptiveExtension();
return compiler.compile(code, classLoader);
}
//看看compiler
@SPI("javassist")
public interface Compiler {
/**
* Compile java source code.
*
* @param code Java source code
* @param classLoader classloader
* @return Compiled class
*/
Class<?> compile(String code, ClassLoader classLoader);
}
//其實到這裡就知道了,通過生成一個類的字元串,再通過javassist生成一個對象 createAdaptiveExtensionClassCode()方法中使用一個StringBuilder來建構自适應類的Java源碼。方法實作比較長，這裡就不貼代碼了。這種生成位元組碼的方式也挺有意思的，先生成Java源代碼，然後編譯，加載到jvm中。通過這種方式，可以更好的控制生成的Java類。而且這樣也不用care各個位元組碼生成架構的api等。因為xxx.java檔案是Java通用的，也是我們最熟悉的。隻是代碼的可讀性不強，需要一點一點建構xx.java的内容。
服務暴露與發現
服務暴露
名詞解釋
在Dubbo 的核心領域模型中：
- Invoker 是實體域，它是Dubbo 的核心模型，其它模型都向它靠擾，或轉換成它，它代表一個可執行體，可向它發起invoke 調用，它有可能是一個本地的實作，也可能是一個遠端的實作，也可能一個叢集實作。在服務提供方，Invoker用于調用服務提供類。在服務消費方，Invoker用于執行遠端調用。
-
Protocol 是服務域，它是Invoker 暴露和引用的主功能入口，它負責Invoker 的生命周期管理。
export：暴露遠端服務
refer：引用遠端服務
-
proxyFactory：擷取一個接口的代理類
getInvoker：針對server端，将服務對象，如DemoServiceImpl包裝成一個Invoker對象
getProxy：針對client端，建立接口的代理對象，例如DemoService的接口。
- Invocation 是會話域，它持有調用過程中的變量，比如方法名，參數等
整體流程
在詳細探讨服務暴露細節之前 ， 我們先看一下整體duubo的服務暴露原理
在整體上看，Dubbo 架構做服務暴露分為兩大部分 ， 第一步将持有的服務執行個體通過代理轉換成Invoker, 第二步會把Invoker 通過具體的協定 （ 比如Dubbo ） 轉換成Exporter, 架構做了這層抽象也大大友善了功能擴充 。
服務提供方暴露服務的藍色初始化鍊，時序圖如下：
服務導出的入口方法是ServiceBean 的onApplicationEvent。onApplicationEvent 是一個事件響應方法，該方法會在收到Spring 上下文重新整理事件後執行服務導出操作。方法代碼如下：
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
if (!isExported() && !isUnexported()) {
if (logger.isInfoEnabled()) {
logger.info("The service ready on spring started. service: " + getInterface());
}
export();
}
} 通過export最終找到doExportUrls()方法
private void doExportUrls() {
//加載配置檔案中的所有注冊中心，并且封裝為dubbo内部的URL對象清單
List<URL> registryURLs = loadRegistries(true);
//循環所有協定配置，根據不同的協定，向注冊中心中發起注冊
for (ProtocolConfig protocolConfig : protocols) {
String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version);
ProviderModel providerModel = new ProviderModel(pathKey, ref, interfaceClass);
ApplicationModel.initProviderModel(pathKey, providerModel);
//服務暴露方法
doExportUrlsFor1Protocol(protocolConfig, registryURLs);
}
} doExportUrlsFor1Protocol()方法代碼老多了,我們隻關系核心的地方
...
if (!SCOPE_NONE.equalsIgnoreCase(scope)) {
//本地暴露，将服務資料記錄到本地JVM中
if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
exportLocal(url);
}
//遠端暴露，向注冊中心發送資料
if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
if (!isOnlyInJvm() && logger.isInfoEnabled()) {
logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
}
if (CollectionUtils.isNotEmpty(registryURLs)) {
for (URL registryURL : registryURLs) {
//if protocol is only injvm ,not register
if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
continue;
}
url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY));
URL monitorUrl = loadMonitor(registryURL);
if (monitorUrl != null) {
url = url.addParameterAndEncoded(MONITOR_KEY, monitorUrl.toFullString());
}
if (logger.isInfoEnabled()) {
logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
}
// For providers, this is used to enable custom proxy to generate invoker
String proxy = url.getParameter(PROXY_KEY);
if (StringUtils.isNotEmpty(proxy)) {
registryURL = registryURL.addParameter(PROXY_KEY, proxy);
}
// 為服務提供類(ref)生成 Invoker
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString()));
// DelegateProviderMetaDataInvoker 用于持有 Invoker 和ServiceConfig
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
// 導出服務，并生成 Exporter
Exporter<?> exporter = protocol.export(wrapperInvoker);
exporters.add(exporter);
}
} else {
//不存在注冊中心,僅導出服務
....
}
/**
* @since 2.7.0
* ServiceData Store
*/
MetadataReportService metadataReportService = null;
if ((metadataReportService = getMetadataReportService()) != null) {
metadataReportService.publishProvider(url);
}
}
}
this.urls.add(url); 上面代碼根據url 中的scope 參數決定服務導出方式，分别如下：
scope = none，不導出服務
scope != remote，導出到本地
scope != local，導出到遠端
不管是導出到本地，還是遠端。進行服務導出之前，均需要先建立Invoker，這是一個很重要的步驟。是以下面先來分析Invoker 的建立過程。Invoker 是由ProxyFactory 建立而來，Dubbo 預設的ProxyFactory 實作類是JavassistProxyFactory。下面我們到JavassistProxyFactory 代碼中，探索Invoker 的建立過程。如下：
@Override
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// 為目标類建立warpper
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
//建立匿名才invoker對象,并實作doinvoke方法
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName,
Class<?>[] parameterTypes,
Object[] arguments) throws Throwable {
// 調用 Wrapper 的 invokeMethod 方法，invokeMethod 最終會調用目标方法
return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
}
};
} Invoke建立成功之後，接下來我們來看本地導出
/**
* always export injvm
*/
private void exportLocal(URL url) {
URL local = URLBuilder.from(url)
.setProtocol(LOCAL_PROTOCOL) // 設定協定頭為 injvm
.setHost(LOCALHOST_VALUE)//本地ip:127.0.0.1
.setPort(0)
.build();
// 建立 Invoker，并導出服務，這裡的 protocol 會在運作時調用 InjvmProtocol 的export 方法
Exporter<?> exporter = protocol.export(
proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
exporters.add(exporter);
logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry url : " + local);
} exportLocal 方法比較簡單，首先根據URL 協定頭決定是否導出服務。若需導出，則建立一個新的URL并将協定頭、主機名以及端口設定成新的值。然後建立Invoker，并調用InjvmProtocol 的export 方法導出服務。下面我們來看一下InjvmProtocol 的export 方法都做了哪些事情。
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(), exporterMap);
} 如上，InjvmProtocol 的export 方法僅建立了一個InjvmExporter，無其他邏輯。到此導出服務到本地就分析完了。
再看看導出服務到遠端
接下來，我們繼續分析導出服務到遠端的過程。導出服務到遠端包含了服務導出與服務注冊兩個過程。先來分析服務導出邏輯。我們把目光移動到RegistryProtocol 的export 方法上。
@Override
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
// 擷取注冊中心 URL
URL registryUrl = getRegistryUrl(originInvoker);
URL providerUrl = getProviderUrl(originInvoker);
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
//導出服務
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);
// 根據 URL 加載 Registry 實作類，比如 ZookeeperRegistry
final Registry registry = getRegistry(originInvoker);
//擷取已注冊的服務提供者 URL，
final URL registeredProviderUrl = getRegisteredProviderUrl(providerUrl, registryUrl);
ProviderInvokerWrapper<T> providerInvokerWrapper = ProviderConsumerRegTable.registerProvider(originInvoker,
registryUrl, registeredProviderUrl);
//to judge if we need to delay publish
boolean register = registeredProviderUrl.getParameter("register", true);
if (register) {
// 向注冊中心注冊服務
register(registryUrl, registeredProviderUrl);
providerInvokerWrapper.setReg(true);
}
// 向注冊中心進行訂閱 override 資料
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
exporter.setRegisterUrl(registeredProviderUrl);
exporter.setSubscribeUrl(overrideSubscribeUrl);
// 建立并傳回 DestroyableExporter
return new DestroyableExporter<>(exporter);
} 上面代碼看起來比較複雜，主要做如下一些操作：
- 調用doLocalExport 導出服務
- 向注冊中心注冊服務
- 向注冊中心進行訂閱override 資料
- 建立并傳回DestroyableExporter
看看doLocalExport 做了什麼
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {
String key = getCacheKey(originInvoker);
return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(key, s -> {
Invoker<?> invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);
//protocol和配置的協定相關（dubbo：DubboProtocol）
return new ExporterChangeableWrapper<>((Exporter<T>) protocol.export(invokerDelegate), originInvoker);
});
} 接下來，我們把重點放在Protocol 的export 方法上。假設運作時協定為dubbo，此處的protocol 變量會在運作時加載DubboProtocol，并調用DubboProtocol 的export 方法。
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
URL url = invoker.getUrl();
// export service.擷取服務辨別，了解成服務坐标也行。由服務組名，服務名，服務版本号以及端口組成。比如：demoGroup/com.alibaba.dubbo.demo.DemoService:1.0.1:20880
String key = serviceKey(url);
//建立DubboExporter
DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
exporterMap.put(key, exporter); //key：接口 （DemoService）
//export an stub service for dispatching event
Boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT);
Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false);
if (isStubSupportEvent && !isCallbackservice) {
String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY);
if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
if (logger.isWarnEnabled()) {
logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) +
"], has set stubproxy support event ,but no stub methods founded."));
}
} else {
stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
}
}
//啟動服務
openServer(url);
//優化序列器
optimizeSerialization(url);
return exporter;
} 如上，我們重點關注DubboExporter 的建立以及openServer 方法，其他邏輯看不懂也沒關系，不影響了解服務導出過程。下面分析openServer 方法。
private void openServer(URL url) {
// find server.
String key = url.getAddress();
//client can export a service which's only for server to invoke
boolean isServer = url.getParameter(IS_SERVER_KEY, true);
if (isServer) {
//通路緩存
ExchangeServer server = serverMap.get(key);
if (server == null) {
synchronized (this) {
server = serverMap.get(key);
if (server == null) {
//建立伺服器執行個體
serverMap.put(key, createServer(url));
}
}
} else {
// server supports reset, use together with override
server.reset(url);
}
}
} 接下來分析伺服器執行個體的建立過程。如下
private ExchangeServer createServer(URL url) {
url = URLBuilder.from(url)
// send readonly event when server closes, it's enabled by default
.addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
// enable heartbeat by default
.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
.addParameter(CODEC_KEY, DubboCodec.NAME)
.build();
String 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
server = Exchangers.bind(url, requestHandler);
} catch (RemotingException e) {
throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
}
// 擷取 client 參數，可指定 netty，mina
str = url.getParameter(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: " + str);
}
}
return server;
} 如上，createServer 包含三個核心的邏輯。
第一是檢測是否存在server 參數所代表的Transporter 拓展，不存在則抛出異常。
第二是建立伺服器執行個體。
第三是檢測是否支援client 參數所表示的Transporter 拓展，不存在也是抛出異常。兩次檢測操作所對應的代碼較直白了，無需多說。但建立伺服器的操作目前還不是很清晰，我們繼續往下看。
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);
} 上面代碼比較簡單，就不多說了。下面看一下HeaderExchanger 的bind 方法。
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
// 建立 HeaderExchangeServer 執行個體，該方法包含了多個邏輯，分别如下：
// 1. new HeaderExchangeHandler(handler)
// 2. new DecodeHandler(new HeaderExchangeHandler(handler))
// 3. Transporters.bind(url, new DecodeHandler(new
HeaderExchangeHandler(handler)))
return new HeaderExchangeServer(Transporters.bind(url, new ChannelHandler[]{new DecodeHandler(new HeaderExchangeHandler(handler))}));
} HeaderExchanger 的bind 方法包含的邏輯比較多，但目前我們僅需關心Transporters 的bind 方法邏
輯即可。該方法的代碼如下：
public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
} else if (handlers != null && handlers.length != 0) {
Object handler;
if (handlers.length == 1) {
handler = handlers[0];
} else {
// 如果 handlers 元素數量大于1，則建立 ChannelHandler 分發器
handler = new ChannelHandlerDispatcher(handlers);
}
// 擷取自适應 Transporter 執行個體，并調用執行個體方法
return getTransporter().bind(url, (ChannelHandler)handler);
} else {
throw new IllegalArgumentException("handlers == null");
}
} 如上，getTransporter() 方法擷取的Transporter 是在運作時動态建立的，類名為TransporterAdaptive，也就是自适應拓展類。TransporterAdaptive 會在運作時根據傳入的URL 參數決定加載什麼類型的Transporter，預設為NettyTransporter。調用 NettyTransporter.bind(URL,ChannelHandler) 方法。建立一個 NettyServer 執行個體。調用 NettyServer.doOPen() 方法，伺服器被開啟，服務也被暴露出來了。
服務注冊
本節内容以Zookeeper 注冊中心作為分析目标，其他類型注冊中心大家可自行分析。下面從服務注冊
的入口方法開始分析，我們把目光再次移到RegistryProtocol 的export 方法上。如下：
進入到register()方法
public void register(URL registryUrl, URL registeredProviderUrl) {
//獲得注冊中心執行個體
Registry registry = registryFactory.getRegistry(registryUrl);
//進行注冊
registry.register(registeredProviderUrl);
} 看看getRegistry()方法
@Override
public Registry getRegistry(URL url) {
url = URLBuilder.from(url)
.setPath(RegistryService.class.getName())
.addParameter(INTERFACE_KEY, RegistryService.class.getName())
.removeParameters(EXPORT_KEY, REFER_KEY)
.build();
String key = url.toServiceStringWithoutResolving();
// Lock the registry access process to ensure a single instance of the registry
LOCK.lock();
try {
Registry registry = REGISTRIES.get(key);
if (registry != null) {
return registry;
}
//create registry by spi/ioc
registry = createRegistry(url);
if (registry == null) {
throw new IllegalStateException("Can not create registry " + url);
}
REGISTRIES.put(key, registry);
return registry;
} finally {
// Release the lock
LOCK.unlock();
}
} 進入createRegistry()方法
@Override
public Registry createRegistry(URL url) {
return new ZookeeperRegistry(url, zookeeperTransporter);
} public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
super(url);
if (url.isAnyHost()) {
throw new IllegalStateException("registry address == null");
}
//// 擷取組名，預設為 dubbo
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);
// 添加狀态監聽器
zkClient.addStateListener(state -> {
if (state == StateListener.RECONNECTED) {
try {
recover();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
});
} 在上面的代碼代碼中，我們重點關注ZookeeperTransporter 的connect 方法調用，這個方法用于建立
Zookeeper 用戶端。建立好Zookeeper 用戶端，意味着注冊中心的建立過程就結束了。
搞懂了服務注冊的本質，那麼接下來我們就可以去閱讀服務注冊的代碼了。
public 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(DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
} @Override
public void create(String path, boolean ephemeral) {
if (!ephemeral) {
// 如果要建立的節點類型非臨時節點，那麼這裡要檢測節點是否存在
if (checkExists(path)) {
return;
}
}
int i = path.lastIndexOf('/');
if (i > 0) {
// 遞歸建立上一級路徑
create(path.substring(0, i), false);
}
// 根據 ephemeral 的值建立臨時或持久節點
if (ephemeral) {
createEphemeral(path);
} else {
createPersistent(path);
}
} 好了，到此關于服務注冊的過程就分析完了。整個過程可簡單總結為：先建立注冊中心執行個體，之後再通過注冊中心執行個體注冊服務。
總結
- 在有注冊中心，需要注冊提供者位址的情況下，ServiceConfig 解析出的URL 格式為：registry:// registry-host/org.apache.dubbo.registry.RegistryService?export=URL.encode("dubbo://service-host/{服務名}/{版本号}")
- 基于Dubbo SPI 的自适應機制，通過URL registry:// 協定頭識别，就調用RegistryProtocol#export() 方法
- 将具體的服務類名，比如 DubboServiceRegistryImpl ，通過ProxyFactory 包裝成Invoker 執行個體
- 調用doLocalExport 方法，使用DubboProtocol 将Invoker 轉化為Exporter 執行個體，并打開Netty 服務端監聽客戶請求
- 建立Registry 執行個體，連接配接Zookeeper，并在服務節點下寫入提供者的URL 位址，注冊服務
- 向注冊中心訂閱override 資料，并傳回一個Exporter 執行個體
- 根據URL 格式中的 "dubbo://service-host/{服務名}/{版本号}" 中協定頭 dubbo:// 識别，調用
DubboProtocol#export()
- RegistryProtocol#export() 傳回的Exporter 執行個體存放到ServiceConfig 的
List<Exporter>exporters